如何在PySpark中使用foreach()操作DataFrame向列表追加元素

你在Driver进程中定义的listOfDfs是Driver本地的变量，foreach、map这类Spark算子的执行逻辑会被序列化后发送到各个Executor节点上运行，Executor操作的是变量的副本，修改不会同步回Driver端，所以最终Driver端的列表始终是空的，和foreach返回空值没有关系。

• 方案1：小数据量场景下直接拉取数据到Driver端处理
  如果DataFrame的数据量不大，不会超出Driver节点的内存上限，可以先把所有行收集到Driver，再遍历处理：

listOfDfs = []
for row in df.collect():
    listOfDfs.extend(getRowInfo(row))
print(listOfDfs)

该方案逻辑最简单，适合测试、小数据集场景。

• 方案2：分布式处理后统一收集结果
  如果需要先分布式执行getRowInfo逻辑，再汇总结果，可以用flatMap算子处理后调用collect拉取结果：

# 先对每一行执行getRowInfo，再将返回的列表展开为单个元素，最后收集到Driver
listOfDfs = df.rdd.flatMap(lambda row: getRowInfo(row)).collect()
print(listOfDfs)

该方案的处理逻辑在分布式节点执行，只在最后一步拉取最终结果到Driver，运行效率更高。

• 方案3：大数据量不需要全量拉回Driver的场景用自定义累加器
  如果最终不需要把全量列表拉回Driver，仅需要在分布式侧做元素追加统计，可以实现Spark自定义累加器完成列表追加，该方案仅适合特定业务场景。

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