Spark 3.2中使用foreachPartition序列化失败问题求助

你遇到的_pickle.PicklingError是因为raw_data_partition函数中直接引用了Driver端的df_cache（即cache_data_test.cache_data）。当Spark把这个函数序列化发送到Worker节点执行时，会尝试序列化整个DataFrame对象，而DataFrame内部持有SparkContext引用——SparkContext只能在Driver端存在，无法被序列化传递到Worker，因此触发报错（对应SPARK-5063的场景）。

核心思路是避免在Worker端直接引用Driver端的DataFrame，改用广播变量将数据分发到Worker节点（广播变量会在每个Worker节点缓存一份，且不会携带SparkContext引用）；如果数据量极小，也可以将DataFrame转为本地集合后传递。

修改后的代码示例 #

def raw_data_partition(partition):
    # 从广播变量中获取缓存数据
    df_cache = broadcast_cache.value
    for row in partition:
        process_row(row, df_cache)

def process_kafka_data(x, epoch_id):
    # 在Driver端将缓存DataFrame转为广播变量
    global broadcast_cache
    broadcast_cache = spark.sparkContext.broadcast(cache_data_test.cache_data.collect())
    x.rdd.foreachPartition(raw_data_partition)

foreachBatch_df = kafka_mapping_df.writeStream \
    .foreachBatch(process_kafka_data) \
    .trigger(processingTime='5 seconds') \
    .option("checkpointLocation", "/home/enc_spark_location") \
    .start()
foreachBatch_df.awaitTermination()

注意事项 #

• 如果cache_data_test.cache_data数据量很大，不建议用collect()转为本地集合后广播，此时可以考虑将缓存数据写入外部存储（比如Redis、Hive），在Worker端的raw_data_partition函数中直接读取外部存储的数据。
• 确保process_row函数中使用的是广播变量的本地数据，而非原始Driver端的DataFrame对象。

内容的提问来源于stack exchange，提问作者Can Yılmaz