升级至Python3.12.2后Spark DataFrame创建报PicklingError求助

问题描述 #

之前使用Python 3.9.6搭配pyspark==3.3.0运行模拟Spark Context的测试代码，创建带Schema的DataFrame一切正常；升级至Python 3.12.2后，执行以下简化测试代码时触发PicklingError异常：

from pyspark.sql import SparkSession
from pyspark.sql.types import (
    StructType,
    StructField,
    StringType,
    TimestampType,
)

BODACC_SCHEMA = StructType(
    [
        StructField("siren", StringType(), True),
        StructField("identifier", StringType(), True),
        StructField("nojo", StringType(), True),
        StructField("type_annonce", StringType(), True),
        StructField("date_publication", TimestampType(), True),
        StructField("source_identifier", StringType(), True),
        StructField("data_source", StringType(), True),
        StructField("created_at", TimestampType(), True),
        StructField("updated_at", TimestampType(), True),
    ]
)


spark_session = SparkSession.builder.master("local[1]").appName("LocalExample").getOrCreate()
spark_session.createDataFrame([], BODACC_SCHEMA)

触发的异常信息 #

Traceback (most recent call last):
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/serializers.py", line 458, in dumps
    return cloudpickle.dumps(obj, pickle_protocol)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 73, in dumps
    cp.dump(obj)
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 602, in dump
    return Pickler.dump(self, obj)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 692, in reducer_override
    return self._function_reduce(obj)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 565, in _function_reduce
    return self._dynamic_function_reduce(obj)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 546, in _dynamic_function_reduce
    state = _function_getstate(func)
            ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle_fast.py", line 157, in _function_getstate
    f_globals_ref = _extract_code_globals(func.__code__)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/cloudpickle/cloudpickle.py", line 334, in _extract_code_globals
    out_names = {names[oparg]: None for _, oparg in _walk_global_ops(co)}
                 ~~~~~^^^^^^^
IndexError: tuple index out of range

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/sql/session.py", line 894, in createDataFrame
    return self._create_dataframe(
           ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/sql/session.py", line 938, in _create_dataframe
    jrdd = self._jvm.SerDeUtil.toJavaArray(rdd._to_java_object_rdd())
                                           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/rdd.py", line 3113, in _to_java_object_rdd
    return self.ctx._jvm.SerDeUtil.pythonToJava(rdd._jrdd, True)
                                                ^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/rdd.py", line 3505, in _jrdd
    wrapped_func = _wrap_function(
                   ^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/rdd.py", line 3362, in _wrap_function
    pickled_command, broadcast_vars, env, includes = _prepare_for_python_RDD(sc, command)
                                                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/rdd.py", line 3345, in _prepare_for_python_RDD
    pickled_command = ser.dumps(command)
                      ^^^^^^^^^^^^^^^^^^
  File "/home/user/.pyenv/versions/3.12.2/envs/data_pipelines/lib/python3.12/site-packages/pyspark/serializers.py", line 468, in dumps
    raise pickle.PicklingError(msg)
_pickle.PicklingError: Could not serialize object: IndexError: tuple index out of range

原因分析 #

1. PySpark 3.3.0与Python 3.12不兼容：Python 3.12对字节码格式、内置函数实现做了变更，而PySpark 3.3.0依赖的旧版cloudpickle无法正确解析Python 3.12生成的函数字节码，导致序列化时出现索引越界错误。PySpark官方从3.4.0版本开始正式支持Python 3.12。
2. debugpy干扰序列化：debugpy会通过修改函数字节码插入调试断点逻辑，旧版PySpark的序列化逻辑无法处理这种被修改后的字节码；PySpark 3.5.1对Python 3.12兼容性更好，但debugpy的字节码修改仍会触发边缘场景的序列化问题，直接运行时无字节码修改所以正常。
3. Dagster+EMR Serverless场景的类似问题：打包过程或EMR Serverless的运行环境可能对代码字节码进行修改/注入（类似debugpy的操作），旧版PySpark的序列化逻辑无法适配这种变更。

修复方案 #

基础问题修复 #

• 方案1：升级PySpark版本：将PySpark升级到3.4.0及以上版本（推荐3.5.1），这些版本官方支持Python 3.12，且依赖的cloudpickle版本已适配Python 3.12的字节码格式。
• 方案2：降级Python版本：如果无法升级PySpark，将Python版本降级到3.11及以下（PySpark 3.3.0支持的最高Python版本为3.11）。

debugpy场景修复 #

• 使用PySpark 3.5+版本，同时调试时避免让debugpy修改需要序列化的Spark相关函数字节码：比如调试时跳过Spark任务的序列化阶段，或者采用远程调试模式且不在Spark核心逻辑中设置断点。

Dagster+EMR Serverless场景修复 #

• 升级PySpark到3.5+版本，确保与Python 3.12兼容。
• 检查Dagster打包配置，禁用可能修改代码字节码的优化选项。
• 确保EMR Serverless使用的Python环境与PySpark版本匹配，避免依赖冲突。

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