如何解读PySpark .explain()及操作顺序？UDF重复执行问题排查

问题说明 #

在PySpark中生成随机二进制数据列后执行过滤操作，多次调用.show()时，结果中频繁出现不符合status != 0过滤条件的数据。推测自定义UDF被执行了两次（过滤前后各一次），但无法解读.explain()的输出内容及阅读顺序，请求分析问题原因。

代码示例 #

from pyspark.sql.types import IntegerType
from pyspark.sql.functions import col, udf
import random
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

_random_udf = udf(lambda x: int(random.randint(0, 1)), IntegerType())

inputDf = spark.createDataFrame([{'row': i} for i in range(10)])
random_result = inputDf.withColumn("status", _random_udf(col("row")))
non_zero_filter = random_result.filter(col('status') != 0)

non_zero_filter.show()

异常输出示例 #

执行计划（.explain()输出） #

== Physical Plan ==
*(3) Project [row#296L, pythonUDF0#314 AS status#299]
+- BatchEvalPython [<lambda>(row#296L)], [pythonUDF0#314]
   +- *(2) Project [row#296L]
      +- *(2) Filter NOT (pythonUDF0#313 = 0)
         +- BatchEvalPython [<lambda>(row#296L)], [pythonUDF0#313]
            +- *(1) Scan ExistingRDD[row#296L]

执行计划解读（阅读顺序从下往上） #

1. *(1) Scan ExistingRDD[row#296L]：最底层操作，读取原始RDD数据源，即我们创建的包含row列的DataFrame。
2. BatchEvalPython [(row#296L)], [pythonUDF0#313]：第一次执行自定义UDF，生成临时列pythonUDF0#313，为后续过滤操作准备数据。
3. *(2) Filter NOT (pythonUDF0#313 = 0)：基于第一次UDF生成的结果执行过滤，保留pythonUDF0#313 != 0的行。
4. *(2) Project [row#296L]：过滤后仅保留row列的数据。
5. BatchEvalPython [(row#296L)], [pythonUDF0#314]：第二次执行自定义UDF，生成新的临时列pythonUDF0#314，为最终展示的status列准备数据。
6. *(3) Project [row#296L, pythonUDF0#314 AS status#299]：将第二次UDF的结果命名为status列，与row列一起输出。

问题原因 #

你的自定义UDF是非确定性的：每次调用random.randint(0,1)都会生成新的随机值。Spark的执行计划中，过滤操作和最终的投影（Project）操作分别触发了一次UDF调用，两次生成的随机值完全独立——过滤时用的是第一次生成的结果（比如某行第一次生成1，通过过滤），但展示时用的是第二次生成的结果（同一行第二次生成0），所以就出现了过滤后结果里有status=0的行。

解决方案 #

方案1：使用Spark内置的随机函数 #

Spark提供了内置的rand()函数，支持指定种子保证确定性，同时可以避免UDF重复执行问题：

from pyspark.sql.functions import rand, when

# 用rand()生成0-1之间的随机数，小于0.5则为0，否则为1，可指定seed保证结果一致
random_result = inputDf.withColumn("status", when(rand() < 0.5, 0).otherwise(1))

如果需要固定随机结果，给rand()传入种子参数，比如rand(seed=42)。

方案2：缓存中间结果 #

在生成包含status列的DataFrame后调用.cache()，强制Spark将中间结果持久化，避免重复计算UDF：

random_result = inputDf.withColumn("status", _random_udf(col("row"))).cache()
non_zero_filter = random_result.filter(col('status') != 0)

注意：缓存后每次启动Spark会话的结果还是随机的，但同一个会话内多次调用.show()会使用缓存的结果，不会出现不符合过滤条件的行。

方案3：将UDF标记为确定性（不推荐用于随机场景） #

如果你的UDF是确定性的（相同输入返回相同输出），可以在定义UDF时指定deterministic=True，但你的场景是生成随机数，这个参数不适用——随机UDF本身就是非确定性的，强制标记会导致结果不符合预期。

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