在 for 循环中将映射函数附加到 PySpark RDD

Question

有人可以帮我理解在 python for 循环中将映射函数附加到 RDD 的行为吗？

对于以下代码：

rdd = spark.sparkContext.parallelize([[1], [2], [3]])

def appender(l, i):
    return l + [i]

for i in range(3):
    rdd = rdd.map(lambda x: appender(x, i))

rdd.collect()

我得到了输出：

[[1, 2, 2, 2], [2, 2, 2, 2], [3, 2, 2, 2]]

而使用以下代码：

rdd = spark.sparkContext.parallelize([[1], [2], [3]])

def appender(l, i):
    return l + [i]

rdd = rdd.map(lambda x: appender(x, 1))
rdd = rdd.map(lambda x: appender(x, 2))
rdd = rdd.map(lambda x: appender(x, 3))

rdd.collect()

我得到了预期的输出：

[[1, 1, 2, 3], [2, 1, 2, 3], [3, 1, 2, 3]]

我想这与传递给 PySpark 编译器的闭包有关，但我找不到任何关于此的文档...

Answer 1

我最好的猜测是因为懒惰的评估： 而且你的射程也很差。

这两个代码 sn-ps 产生相同的输出：

rdd = spark.sparkContext.parallelize([[1], [2], [3]])

def appender(l, i):
    return l + [i]

for i in range(1,4):
    rdd = spark.sparkContext.parallelize(rdd.map(lambda x: appender(x, i)).collect())

rdd.collect()

输出：

[[1, 1, 2, 3], [2, 1, 2, 3], [3, 1, 2, 3]]

第二个：

rdd = spark.sparkContext.parallelize([[1], [2], [3]])

rdd = rdd.map(lambda x: appender(x, 1))
rdd = rdd.map(lambda x: appender(x, 2))
rdd = rdd.map(lambda x: appender(x, 3))

rdd.collect()

输出：

[[1, 1, 2, 3], [2, 1, 2, 3], [3, 1, 2, 3]]

---

另外，为了显示在简化示例（仅输入 1 和 2）中 for 循环中发生的情况，修改了 appender 函数以打印 l 参数：

1. for 循环打印：

   [2]
   [2, 2]
   [1]
   [3]
   [1, 2]
   [3, 2]
   

首先它从输入列表中获取第二个字段

1. 映射器输出的显式写入是：

   [1]
   [1, 1]
   [2]
   [2, 1]
   [3]
   [3, 1]
   

Answer 2

解决方案是将所有全局变量（在本例中为 i）存储在 lambda 函数中，以确保正确关闭。这可以通过

for i in range(3):
    rdd = rdd.map(lambda x, i=i: appender(x, i))

更多信息请访问lambda function accessing outside variable。

有趣的是，至少在本地集群上（还没有在分布式集群上测试过），这个问题也可以通过持久化中间rdd来解决：

for i in range(3):
    rdd = rdd.map(lambda x: appender(x, i))
    rdd.persist()

两种解决方案都产生

[[1, 0, 1, 2], [2, 0, 1, 2], [3, 0, 1, 2]]