python - 将 SparkContext 传递给新进程(python 多处理模块)

Question

我正在使用 Python/Spark 运行一系列不同的作业，一次一个。为了避免每次创建 SparkContex(这需要一段时间)，我想将上下文作为参数发送给每个作业。最重要的是，我希望管理器(创建上下文并运行作业的代码)具有超时机制。

我在第一次作业运行时遇到了一个奇怪的错误，之后它就消失了。

Traceback (most recent call last):
File "/usr/lib/python3.4/multiprocessing/process.py", line 254, in _bootstrap
    self.run()
  File "/usr/lib/python3.4/multiprocessing/process.py", line 93, in run
    self._target(*self._args, **self._kwargs)
  File "/home/dev/ComponentEngine/components/ExampleComponent/ExampleComponent.py", line 35, in run
    numbers = sparkContext.parallelize([1,2,3,4,5,6,7,8,9,10])
  File "/home/dev/Programs/spark-1.4.1-bin-hadoop2.6/python/pyspark/context.py", line 395, in parallelize
    readRDDFromFile = self._jvm.PythonRDD.readRDDFromFile
  File "/home/dev/Programs/spark-1.4.1-bin-hadoop2.6/python/lib/py4j-0.8.2.1-src.zip/py4j/java_gateway.py", line 772, in __getattr__
    raise Py4JError('{0} does not exist in the JVM'.format(name))
py4j.protocol.Py4JError: PythonRDD does not exist in the JVM

代码:

#!/bin/python3
import multiprocessing
from pyspark import SparkContext

def doJob(sc):
    try:
        sc.parallelize([1,2,3,4,5,6,7,8,9,10])
    except Exception as e:
        print('Got excpetion {}'.format(e))

def runWithTimeout(sc):
    p = multiprocessing.Process(target=doJob, name="runWithTimeout", args=(sc))
    p.start()

    # Wait till the timeout
    p.join(10)

    if p.is_alive():
        p.terminate()
        p.join()


if __name__ == '__main__':
    sc = SparkContext()
    for i in range(3):
        runWithTimeout(sc)

为什么会出现这个错误？
这样传递SparkContext有什么问题吗？我知道它会被另一端的进程序列化并使用，并且如果作业对上下文执行任何更改，引擎的副本不会受到影响。
是否有任何此类更改会干扰其他作业的运行？

Answer 1

序列化 Spark 上下文在任何受支持的语言中都无法正常工作。一般来说，所做的事情类似于 IBM Spark Kernel 或 ooyola jobserver，其中单个进程* 保存 Spark 上下文，并且多个客户端与服务器通信。在 python 中，Spark Context 包括用于与 JVM SparkContext 通信的网络套接字，而网络套接字并不是真正可序列化的对象。看看 py4j(Spark 用于在 python 和 JVM 之间进行通信的库)，多线程可以工作，因为它们可以共享套接字，但多个进程则不然。