PySpark在ForEachBatch中循环执行转换报SparkContext仅可在Driver创建错误

实现需求 #

从Python字典中读取统一的数据转换规则，在Structured Streaming的foreachBatch方法中按规则依次转换后输出结果数据，所有转换方法均满足输入为DataFrame、输出为DataFrame的格式要求。

基础代码实现 #

writeStream调用逻辑 #

pipeline.writeStream \
        .queryName(self.pipeline_name) \
        .foreachBatch(self.transformation_chain()) \
        .start()

初始固定转换逻辑 #

def transformation_chain(self):
    def chain(df, epoch_id):
        df_in_edit = df
        df_in_edit = filter_out_null(df_in_edit)
        df_in_edit = split_one_stream_into_two(df_in_edit, "direction", "velocity")
        #for activity in self.data_transformations_workflow:
        #    df_in_edit = activity["transformation"](df_in_edit, *activity["args"])

        df_in_edit.coalesce(1).write.save(path=self.output_path, format=self.output_format, mode='append')

    return chain

通用转换规则配置 #

为了实现可配置的转换流程，定义了转换规则列表：

data_transformations_workflow = [                              
                                {
                                  "transformation": filter_out_sensor_columns_presence_validation,
                                  "args": ("value", "direction velocity")
                                },
                                {
                                  "transformation": split_one_stream_into_two,
                                  "args": ("direction", "velocity")
                                }
                            ]

迭代调用转换的改造代码 #

def transformation_chain(self):
    def chain(df, epoch_id):
        df_in_edit = df
        for activity in self.data_transformations_workflow:
            df_in_edit = activity["transformation"](df_in_edit, *activity["args"])

        df_in_edit.coalesce(1).write.save(path=self.output_path, format=self.output_format, mode='append')

    return chain

报错信息 #

改造后运行触发如下错误：

WARNING: An illegal reflective access operation has occurred
WARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/usr/local/spark-3.1.2-bin-hadoop3.2/jars/spark-unsafe_2.12-3.1.2.jar) to constructor java.nio.DirectByteBuffer(long,int)
WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.Platform
WARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operations
WARNING: All illegal access operations will be denied in a future release
:: loading settings :: url = jar:file:/usr/local/spark-3.1.2-bin-hadoop3.2/jars/ivy-2.4.0.jar!/org/apache/ivy/core/settings/ivysettings.xml
Ivy Default Cache set to: /home/jovyan/.ivy2/cache
The jars for the packages stored in: /home/jovyan/.ivy2/jars
org.apache.spark#spark-sql-kafka-0-10_2.12 added as a dependency
:: resolving dependencies :: org.apache.spark#spark-submit-parent-30b1f701-8caa-47db-b5fe-5408ccfe09b2;1.0
    confs: [default]
    found org.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 in central
    found org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 in central
    found org.apache.kafka#kafka-clients;2.4.1 in central
    found com.github.luben#zstd-jni;1.4.4-3 in central
    found org.lz4#lz4-java;1.7.1 in central
    found org.xerial.snappy#snappy-java;1.1.7.5 in central
    found org.slf4j#slf4j-api;1.7.30 in central
    found org.spark-project.spark#unused;1.0.0 in central
    found org.apache.commons#commons-pool2;2.6.2 in central
:: resolution report :: resolve 697ms :: artifacts dl 13ms
    :: modules in use:
    com.github.luben#zstd-jni;1.4.4-3 from central in [default]
    org.apache.commons#commons-pool2;2.6.2 from central in [default]
    org.apache.kafka#kafka-clients;2.4.1 from central in [default]
    org.apache.spark#spark-sql-kafka-0-10_2.12;3.0.0 from central in [default]
    org.apache.spark#spark-token-provider-kafka-0-10_2.12;3.0.0 from central in [default]
    org.lz4#lz4-java;1.7.1 from central in [default]
    org.slf4j#slf4j-api;1.7.30 from central in [default]
    org.spark-project.spark#unused;1.0.0 from central in [default]
    org.xerial.snappy#snappy-java;1.1.7.5 from central in [default]
    ---------------------------------------------------------------------
    |                  |            modules            ||   artifacts   |
    |       conf       | number| search|dwnlded|evicted|| number|dwnlded|
    ---------------------------------------------------------------------
    |      default     |   9   |   0   |   0   |   0   ||   9   |   0   |
    ---------------------------------------------------------------------
:: retrieving :: org.apache.spark#spark-submit-parent-30b1f701-8caa-47db-b5fe-5408ccfe09b2
    confs: [default]
    0 artifacts copied, 9 already retrieved (0kB/16ms)
21/08/18 09:40:43 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust log level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
/usr/local/spark/python/lib/pyspark.zip/pyspark/shell.py:42: UserWarning: Failed to initialize Spark session.
Traceback (most recent call last):
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/shell.py", line 38, in <module>
    spark = SparkSession._create_shell_session()  # type: ignore
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 543, in _create_shell_session
    return SparkSession.builder\
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/sql/session.py", line 228, in getOrCreate
    sc = SparkContext.getOrCreate(sparkConf)
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/context.py", line 384, in getOrCreate
    SparkContext(conf=conf or SparkConf())
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/context.py", line 136, in __init__
    SparkContext._assert_on_driver()
  File "/usr/local/spark/python/lib/pyspark.zip/pyspark/context.py", line 1277, in _assert_on_driver
    raise Exception("SparkContext should only be created and accessed on the driver.")
Exception: SparkContext should only be created and accessed on the driver.

问题现象 #

每次循环执行转换规则时都会触发Spark重新初始化，仅当for循环后执行CSV写入操作时才会触发报错，将写入操作放到for循环之前可正常运行，但不符合业务逻辑需求。确认未在Executor中直接使用SparkContext。

排查更新 #

更新1：局部变量测试 #

将转换规则列表改为chain方法内的局部变量，问题仍未复现，核心类定义如下：

class DataPipelineGenerator:
    pipeline_name = "pipeline"
    subscribed_topic = "subscribed"
    data_transformations_workflow = [                              
                                        {
                                          "transformation": filter_out_sensor_columns_presence_validation,
                                          "args": ("value", "direction velocity")
                                        },
                                        {
                                          "transformation": split_one_stream_into_two,
                                          "args": ("direction", "velocity")
                                        }
                                    ]
    output_path = "/home/jovyan/work/notebooks/pipelines/"
    output_format = "csv"
    read_stream = []

    def __init__(self, pipeline_name, subscribed_topic, data_transformations_workflow, output_path, output_format,spark):
        self.pipeline_name = pipeline_name
        self.subscribed_topic = subscribed_topic
        self.output_path = output_path
        self.output_format = output_format
        self.read_stream = self.init_read_stream(spark)

    def init_read_stream(self, spark):
        return spark \
            .readStream \
            .format('kafka') \
            .option('kafka.bootstrap.servers', bootstrap_servers) \
            .option("startingOffsets", "earliest") \
            .option('subscribe', self.subscribed_topic) \
            .load() \
            .selectExpr('CAST(key AS STRING)', 'CAST(value AS STRING)') \
            .select(json_tuple(col("value"), "ts", "key", "value")) \
            .selectExpr("c0 as ts", "c1 as key", "c2 as value") \
            .withColumn("key_temp", expr("substring(key, {}, length(key))".format(clipping_prefix))) \
            .drop("key") \
            .withColumnRenamed("key_temp", "key")

    def init_write_stream(self):
        return self.read_stream.writeStream \
            .queryName(self.pipeline_name) \
            .foreachBatch(self.transformation_chain()) \
            .start()

    def transformation_chain(self):
        def chain(df, epoch_id):
            df_in_edit = df
            data_transformations_workflow_local = [
                {
                    "transformation": filter_out_sensor_columns_presence_validation,
                    "args": ("value", "direction velocity")
                },
                {
                    "transformation": split_one_stream_into_two,
                    "args": ("direction", "velocity")
                }
            ]
            for activity in data_transformations_workflow_local:
                df_in_edit = activity["transformation"](df_in_edit, *activity["args"])

            df_in_edit.coalesce(1).write.save(path=self.output_path, format=self.output_format, mode='append')

        return chain

更新2：根源定位 #

最终发现问题根源为：转换方法所在的Python文件中存在全局调用emp_RDD = spark.sparkContext.emptyRDD()，每次导入/调用该文件中的方法时都会尝试创建新的SparkContext，目前尚未明确单独调用方法未触发该问题的具体原因。 #

内容的提问来源于stack exchange，提问作者LikeAKemper