Spark Filter算子异常行为：变量引用导致结果不符合预期

这是个很经典的Spark + Python结合时容易踩的坑，核心是Spark RDD的惰性求值特性和Python闭包的延迟绑定机制叠加导致的，我来给你拆解清楚：

问题本质：两个特性的叠加效应 #

1. Spark RDD的惰性求值与Lineage依赖
   RDD的转换操作（比如filter）不会立即执行计算，只是记录下数据处理的依赖链（也就是Lineage）。只有当遇到行动操作（比如collect）时，才会从头触发整个依赖链的计算。而且如果RDD没有被缓存，每次执行行动操作都会重新计算一次。

2. Python闭包的延迟绑定
   你在lambda里引用的变量t，并不是在定义lambda时就固定它的当前值，而是在lambda实际执行的时候才去查找变量的最新值。

原代码的执行流程拆解 #

我们一步步看你的代码到底发生了什么：

A = sc.parallelize(xrange(1, 100))
t = 50
B = A.filter(lambda x: x < t)  # 这里只是记录过滤逻辑，没有计算，lambda记住的是变量t的引用
print B.collect()  # 触发行动操作，此时t的值是50，所以B计算出1-49，正常打印
t = 10  # 修改t的值为10
C = B.filter(lambda x: x > t)  # 同样只是记录逻辑，lambda仍然记住t的引用
print C.collect()  # 触发行动操作：
                   # 1. 因为B没有被缓存，需要重新计算B
                   # 2. 此时执行B的lambda，查找t的当前值是10，所以B的结果变成1-9
                   # 3. 再执行C的lambda，过滤x>10，自然没有符合条件的元素，返回空数组

为什么用新变量m就正常？ #

当你改用新变量m=10时，代码中并没有修改原来的t的值（t始终是50）：

A = sc.parallelize(xrange(1, 100))
t = 50
B = A.filter(lambda x: x < t)
print B.collect()
m = 10  # 定义新变量m，t仍然保持50不变
C = B.filter(lambda x: x > m)
print C.collect()  # 触发行动操作：
                   # 1. 重新计算B时，t的值还是50，所以B的结果是1-49
                   # 2. 执行C的lambda时，m的值是10，过滤出11-49，结果正常

两种可行的解决方案 #

如果你想保留使用变量t的写法，有两种常用的解决方法：

1. 缓存RDD B
   在第一次计算B之后，调用B.cache()把结果缓存起来，后续计算C时就会直接使用缓存的结果，不会重新计算B：

   A = sc.parallelize(xrange(1, 100))
   t = 50
   B = A.filter(lambda x: x < t)
   print B.collect()
   B.cache()  # 缓存B的计算结果
   t = 10
   C = B.filter(lambda x: x > t)
   print C.collect()  # 此时使用缓存的B（1-49），过滤x>10得到11-49
   

2. 定义lambda时强制捕获变量当前值
   通过默认参数的方式，让lambda在定义时就固定变量的当前值，避免延迟引用：

   A = sc.parallelize(xrange(1, 100))
   t = 50
   # 用默认参数t=t，强制捕获当前t的值（50）
   B = A.filter(lambda x, t=t: x < t)
   print B.collect()
   t = 10
   # 同样用默认参数捕获当前t的值（10）
   C = B.filter(lambda x, t=t: x > t)
   print C.collect()  # B的lambda已经绑定了t=50，所以B结果是1-49，C过滤x>10得到正确结果
   

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