PySpark RDD filter条件变量修改后执行行为异常问题咨询

核心原因 #

你遇到的问题是Spark惰性求值和Python闭包后期绑定两个特性共同作用的结果：

• Spark RDD的转换操作（如filter）是惰性执行的，定义时不会实际计算，只会记录依赖血缘，只有遇到行动操作（如count、collect）时才会触发全链路计算。
• Python中lambda闭包引用的自由变量是后期绑定的，变量值不会在lambda定义时固定，而是在lambda实际执行时才读取当前值。

第一段代码结果解释 #

你第一段代码的执行逻辑符合上述特性：

1. 定义B = A.filter(lambda x: x < t)时，只是记录了过滤规则，没有实际计算。
2. 第一次B.count()触发计算，此时t=50，过滤条件为x<50，得到结果49。
3. 修改t=10后执行C.count()，需要重新从A开始计算整条血缘，此时B的过滤条件读取到的t已经是10，B的结果为[1,2,...,9]，再叠加C的过滤条件x>10，最终结果为空，输出0。

第二段代码奇怪输出解释 #

你观察到的第二次B.collect()返回1-49、但B.count()返回9的现象，是本地测试环境的特殊表现，不属于Spark标准行为：

• 本地模式下小数据量的collect结果会被Python侧临时缓存，第二次调用collect时没有重新触发RDD计算，直接返回了第一次执行的结果。
• 而count操作触发了新一轮全量计算，此时读取到的t=10，过滤后结果只有9个元素，输出9。
  如果在分布式集群环境测试，你会看到第二次B.collect()同样返回[1,2,...,9]，和count结果一致。

解决方案 #

有两种常用方法可以避免这类闭包变量异常变更的问题：

1. 利用Python默认参数的早期绑定特性，定义lambda时就固定变量值：

B = A.filter(lambda x, t=t: x < t)

2. 如果需要复用中间RDD的计算结果，直接对中间RDD做缓存：

B = A.filter(lambda x: x < t).cache()
# 第一次action计算后结果会持久化到内存，后续操作不会重新计算血缘
print(B.count())

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