PySpark中传入MapPartitions的函数被调用8次问题求助

核心成因 #

该问题由Spark的核心机制和Databricks的默认环境配置共同导致：

• Spark默认采用惰性求值策略，所有转换算子仅记录计算逻辑，不会立即执行；未做持久化的算子链路，每次遇到行动算子都会从头重新执行全链路逻辑。
• 代码中共显式触发2次行动算子：resultsDF.collect()、sdf.count()，单独这两次操作就会触发func_1执行2次。
• 剩余6次调用来自Databricks的隐式校验逻辑：调用spark.createDataFrame时，即便提前指定了schema，Databricks默认会执行多轮数据采样校验，确认数据格式和schema匹配；若环境开启了自适应执行、元数据预校验、小数据量预览等默认配置，也会额外触发RDD计算采样，这部分隐式触发的调用合计6次，叠加显式触发的2次，最终出现func_1被调用8次的现象。

修复方案 #

对依赖func_1生成的中间结果做缓存持久化，即可避免重复执行：

def func_1(*args, **kwargs):
     print("some work here")
     yield [Row('field1', 'field2')(1, 2)]

test_list = [[1, 2, 3, 4, 5]]
testRDD = sc.parallelize(test_list, 1)
# mapPartitions后添加cache持久化计算结果
resultRDD  = testRDD.mapPartitions(func_1, 1).cache()
resultsDF = spark.createDataFrame(resultRDD.flatMap(lambda x: x), schema=someSchema) 
resultsDF.collect()
sdf = sdf.union(resultsDF)
print(sdf.count())
# 后续不再使用缓存时可主动释放资源
resultRDD.unpersist()

缓存后func_1仅会在第一次行动算子触发时执行1次，后续所有操作都会直接读取缓存结果，不会重复执行func_1逻辑。

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