如何使用map函数并行运行PySpark代码并解决PicklingError序列化报错

你触发序列化错误的核心原因是：你将fullData这个DataFrame对象传入了RDD的map算子中。Spark需要将算子内用到的所有对象序列化后分发到Executor节点执行，而DataFrame绑定了SparkSession上下文，内部包含_thread.RLock这类不可序列化的线程锁对象，因此会触发pickle序列化失败。
另外你原有的实现逻辑即使解决了序列化问题，也存在严重的性能缺陷：每个分片任务都需要扫描全量fullData做过滤，数据量大时执行效率极低，且返回的entries仍是分布式DataFrame对象，collect回Driver后也无法直接使用。

直接使用Spark原生的groupBy分组算子实现需求，完全规避序列化问题同时最大化执行效率：

from pyspark.sql.functions import collect_list, concat, lit, regexp_replace, col

def main(fullData):
    # 按两个主键分组，收集经纬度列表，同时生成目标文件名
    result_df = fullData.groupBy("vehicle_id", "start_time") \
        .agg(
            collect_list("latitude").alias("latitude_list"),
            collect_list("longitude").alias("longitude_list")
        ) \
        .withColumn("folder", concat(lit("/playback/"), col("vehicle_id"), lit("/"))) \
        .withColumn("file_name", concat(
            col("folder"),
            regexp_replace(regexp_replace(col("start_time"), " ", "_"), ":", "-")
        )) \
        .select("file_name", "latitude_list", "longitude_list")
    
    # 如果需要回收到Driver本地处理，可以执行collect，否则直接用DataFrame做后续存储等操作
    jsons = result_df.collect()
    return jsons

方案优势 #

• 完全避免了向分布式算子中传递不可序列化的上下文对象，不会触发序列化错误
• 依赖Spark原生优化的分组shuffle逻辑执行，比逐次过滤全量数据的写法性能高数十倍，支持TB级以上数据量处理
• 所有计算全分布式执行，不存在Driver单点性能瓶颈，可根据集群资源水平扩展
• 生成的结果可直接落地存储，无需额外处理DataFrame嵌套的问题

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