PySpark结构化流序列化失败：无法pickle '_thread.lock'对象

报错根源 #

你遇到的_pickle.PicklingError是因为Spark在分布式执行foreachPartition时，需要将传递给Executor的对象序列化。你的ProcessActions类实例中包含了无法被pickle序列化的对象——比如logger（内部带有_thread.lock线程锁）、spark_session（SparkSession本身不能跨节点序列化），当Spark尝试把类实例连同process_rows方法一起序列化发送到Executor时，就会触发这个错误。

而之前用df.collect()没问题，是因为collect()会把所有数据拉到Driver节点处理，所有逻辑都在本地执行，不需要序列化类实例到Executor，自然不会触发序列化问题，但这种做法确实会丢失Spark的并行处理能力。

解决方法 #

核心思路是：避免将不可序列化的对象传递到Executor，只把必要的可序列化配置分发到Executor，在Executor本地初始化需要的资源。

方案1：重构类逻辑，剥离不可序列化成员 #

修改ProcessActions类，只保留可序列化的配置参数，在Executor端的process_rows内部初始化日志、Spark相关资源：

class ProcessActions:
    def __init__(self, config, rule):
        # 仅保存可序列化的配置数据，去掉logger、spark_session
        self.config = config
        self.rule = rule

    def process(self, df: pyspark.sql.DataFrame, _: int) -> None:
        if df.isEmpty():
            return
        
        # 用广播变量分发可序列化配置（比直接传递更高效）
        bc_config = df.sparkSession.sparkContext.broadcast(self.config)
        bc_rule = df.sparkSession.sparkContext.broadcast(self.rule)

        def process_rows(partition):
            # 在Executor本地初始化日志工具
            import logging
            logger = logging.getLogger(f"process_{bc_rule.value['module_name']}")
            logger.setLevel(logging.INFO)
            
            # 获取广播的配置
            config = bc_config.value
            rule = bc_rule.value

            # 处理当前分区的每一行数据
            for row in partition:
                # 这里替换成你的业务逻辑
                logger.info(f"Processing row: {row}")

        df.foreachPartition(process_rows)

方案2：使用独立函数替代类方法 #

如果不需要类封装，也可以把处理逻辑改成独立函数，直接传递可序列化参数：

def process_batch(df: pyspark.sql.DataFrame, batch_id: int, config, rule):
    if df.isEmpty():
        return
    
    bc_config = df.sparkSession.sparkContext.broadcast(config)
    bc_rule = df.sparkSession.sparkContext.broadcast(rule)

    def process_rows(partition):
        import logging
        logger = logging.getLogger(f"process_{bc_rule.value['module_name']}")
        # 业务逻辑...
        for row in partition:
            pass

    df.foreachPartition(process_rows)

# 在writeStream中调用时：
streaming_query = (
    df.writeStream.option(
        "checkpointLocation",
        f'{config["checkpoint_location"]}/{rule["module_name"]}',
    )
    .foreachBatch(lambda df, bid: process_batch(df, bid, config, rule))
    .trigger(processingTime="0 seconds")
    .queryName(rule["module_name"])
    .start()
)

关键注意点 #

• 永远不要把SparkSession、Logger（带线程锁的实例）、数据库连接池这类不可序列化的对象传递给Executor，必须在Executor本地初始化。
• 广播变量适合分发大体积的可序列化配置，能减少网络传输开销。
• foreachPartition的处理函数会在Executor的每个任务进程中执行，所以内部初始化的资源（比如logger）是进程级的，无需担心线程安全问题。

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