在 Spark 独立和 YARN 客户端部署模式上运行时，cythonize 代码如何工作，但在 YARN 集群部署上却不行？答案

【问题标题】：How come cythonize code works when running on Spark standalone and YARN client deployment mode, but not on YARN cluster deployment?在 Spark 独立和 YARN 客户端部署模式上运行时，cythonize 代码如何工作，但在 YARN 集群部署上却不行？
【发布时间】：2021-11-19 08:17:02
【问题描述】：

我有一个 Python 库，我有一个使用 this approach 的 cythonized。在我的所有 Spark 集群节点上，我都安装了 whl 文件，如下所示。

pip install myapi-0.0.1-cp38-cp38-linux_x86_64.whl

当我按如下方式向 Spark 独立提交作业时，代码运行良好。

spark-submit \
    --master spark://172.18.0.32:7077 \
    test.py

当我使用客户端部署模式通过 YARN 提交作业时，代码也运行良好。

spark-submit \
    --master yarn \
    --deploy-mode client \
    test.py

但是，当我通过 YARN 集群部署模式提交作业时，代码会中断。

spark-submit \
    --master yarn \
    --deploy-mode cluster \
    test.py

特别是，我收到此错误。

pickle.PicklingError: Can't pickle : 在 myapi 上查找属性 lambda。

代码myapi.utils.Data 没什么特别的，看起来像这样。

class Data:
    def __init__(self, rdd):
        self.rdd = rdd

    def compute(self):
        return self.rdd.map(change_it).reduce(lambda a, b: a + b)

def change_it(n):
    a = lambda v: v
    b = lambda v: v
    c = lambda v: v
    d = lambda v: v
    e = lambda v: v
    f = lambda v: a(b(c(d(e(v)))))
    
    return f(n)

few places 讨论了使用 PySpark、pickle、cloudpickle 和 cythonized 模块酸洗（Web 和 SO）嵌套函数的困难。但是，答案似乎不是blanket 的答案，无法解释为什么它在一种情况下有效，而在我上面实验过的其他情况下无效。

任何关于我为什么得到这些观察结果的进一步解释将不胜感激。

我使用 Spark 的环境设置如下。

Spark v3.3.1
Hadoop v3.2.1
Python v3.8

【问题讨论】：

标签： apache-spark pyspark cython cloudpickle

【解决方案1】：

我对 Spark 不是很熟悉，但我希望“客户端”模式在同一个 Python 进程中运行，因此不需要序列化任何内容（可以简单地创建和使用对象）。 “集群”模式可能被设计为在一堆不同的计算机上运行，因此需要传递给经过腌制和解封的单独进程的数据才能分发。

我相信 cloudpickle 和 dill 可以查看常规 Python 函数和 lambda，提取字节码，然后重构函数。这对于 Cython 显然是不可能的。在当前版本的 Cython 中，函数是按名称腌制的（因此 lambdas 不是可腌制的，也不是任何内部函数）。

您的选择是：

重写任何根据可腌制类捕获变量的函数或 lambda（使用 __call__ 函数）；用在函数或类范围内定义的 def 函数替换其他 lambda。
试试this experimental branch，它可以让大部分 Cython 函数都可以腌制。

【讨论】：