如何拆除 SparkSession 并在一个应用程序中创建一个新的？

Question

我有一个带有多个独立模块的 pyspark 程序，每个模块可以独立处理数据以满足我的各种需求。但它们也可以链接在一起以处理管道中的数据。这些模块中的每一个都构建了一个 SparkSession 并自行完美执行。

但是，当我尝试在同一个 python 进程中连续运行它们时，我遇到了问题。在管道中的第二个模块执行的那一刻，spark 抱怨我尝试使用的 SparkContext 已停止：

py4j.protocol.Py4JJavaError: An error occurred while calling o149.parquet.
: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.

这些模块中的每一个都在执行开始时构建一个 SparkSession，并在其进程结束时停止 sparkContext。我像这样构建和停止会话/上下文：

session = SparkSession.builder.appName("myApp").getOrCreate()
session.stop()

根据official documentation，getOrCreate“获取现有的 SparkSession，或者，如果没有现有的，则根据此构建器中设置的选项创建一个新的。”但我不想要这种行为（进程试图获取现有会话的这种行为）。我找不到任何方法来禁用它，也无法弄清楚如何销毁会话——我只知道如何停止其关联的 SparkContext。

如何在独立的模块中构建新的 SparkSession，并在同一个 Python 进程中按顺序执行它们，而以前的会话不会干扰新创建的会话？

以下是项目结构示例：

ma​​in.py

import collect
import process

if __name__ == '__main__':
    data = collect.execute()
    process.execute(data)

collect.py

import datagetter

def execute(data=None):
    session = SparkSession.builder.appName("myApp").getOrCreate()

    data = data if data else datagetter.get()
    rdd = session.sparkContext.parallelize(data)
    [... do some work here ...]
    result = rdd.collect()
    session.stop()
    return result

process.py

import datagetter

def execute(data=None):
    session = SparkSession.builder.appName("myApp").getOrCreate()
    data = data if data else datagetter.get()
    rdd = session.sparkContext.parallelize(data)
    [... do some work here ...]
    result = rdd.collect()
    session.stop()
    return result

Answer 1

长话短说，Spark（包括 PySpark）并非旨在在单个应用程序中处理多个上下文。如果您对 JVM 方面的故事感兴趣，我建议您阅读SPARK-2243（已解决为无法修复）。

在 PySpark 中做出了许多反映这一点的设计决策，包括但不限于 a singleton Py4J gateway。有效you cannot have multiple SparkContexts in a single application。 SparkSession 不仅绑定到 SparkContext，而且还引入了它自己的问题，比如处理本地（独立）Hive 元存储（如果使用的话）。此外，还有在内部使用SparkSession.builder.getOrCreate 的函数，取决于你现在看到的行为。一个值得注意的例子是 UDF 注册。如果存在多个 SQL 上下文（例如 RDD.toDF），其他函数可能会出现意外行为。

多个上下文不仅不受支持，而且在我个人看来，这违反了单一责任原则。您的业​​务逻辑不应该关心所有设置、清理和配置细节。

我的个人建议如下：

• 如果应用程序由多个连贯的模块组成，这些模块可以组合并受益于具有缓存和公共元存储的单个执行环境，则在应用程序入口点初始化所有必需的上下文，并在必要时将它们传递到各个管道：

  • main.py:

    from pyspark.sql import SparkSession
    
    import collect
    import process
    
    if __name__ == "__main__":
        spark: SparkSession = ...
    
        # Pass data between modules
        collected = collect.execute(spark)
        processed = process.execute(spark, data=collected)
        ...
        spark.stop()
    

  • collect.py/process.py:

    from pyspark.sql import SparkSession
    
    def execute(spark: SparkSession, data=None):
        ...
    

• 否则（根据您的描述，这里似乎是这种情况）我会设计入口点来执行单个管道并使用外部工作流管理器（如Apache Airflow 或Toil）来处理执行。

  它不仅更干净，而且还允许更灵活的故障恢复和调度。

  同样的事情当然也可以用建设者来完成，但就像 smart person 曾经说过的那样：显式优于隐式。

  • main.py

    import argparse
    
    from pyspark.sql import SparkSession
    
    import collect
    import process
    
    pipelines = {"collect": collect, "process": process}
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser()
        parser.add_argument('--pipeline')
        args = parser.parse_args()
    
        spark: SparkSession = ...
    
        # Execute a single pipeline only for side effects
        pipelines[args.pipeline].execute(spark)
        spark.stop()
    

  • collect.py / process.py 同上一点。

我会以一种或另一种方式保留一个且只有一个设置上下文的位置，以及一个且只有一个要拆除的位置。

Answer 2

这是一种解决方法，但不是解决方案：

我发现source code 中的SparkSession 类包含以下__init__（我已从此处删除了不相关的代码行）：

_instantiatedContext = None

def __init__(self, sparkContext, jsparkSession=None):
    self._sc = sparkContext
    if SparkSession._instantiatedContext is None:
        SparkSession._instantiatedContext = self

因此，我可以通过在调用session.stop() 后将会话的_instantiatedContext 属性设置为None 来解决我的问题。当下一个模块执行时，它调用getOrCreate()，并没有找到之前的_instantiatedContext，所以它分配了一个新的sparkContext。

这不是一个非常令人满意的解决方案，但它可以作为一种解决方法来满足我当前的需求。我不确定这种开始独立会话的整个方法是反模式还是不寻常。

Answer 3

为什么不将同一个 Spark 会话实例传递到管道的多个阶段？您可以使用构建器模式。在我看来，您在每个阶段的最后都收集结果集，然后将这些数据传递到下一个阶段。考虑将集群中的数据留在同一个会话中，并将会话引用和结果引用从一个阶段传递到另一个阶段，直到您的应用程序完成。

换句话说，把

session = SparkSession.builder...

...在您的主目录中。

Answer 4

spark_current_session = SparkSession. \
                             builder. \
                             appName('APP'). \
                             config(conf=SparkConf()). \
                             getOrCreate()
spark_current_session.newSession()

您可以从当前会话创建一个新会话