如何从 PySpark DataFrame 中批量处理项目答案

【问题标题】：How to batch up items from a PySpark DataFrame如何从 PySpark DataFrame 中批量处理项目
【发布时间】：2019-09-22 13:04:06
【问题描述】：

我有一个 PySpark 数据框，对于每条（一批）记录，我想调用一个 API。所以基本上说我有 100000k 记录，我想将项目分成 1000 个组并调用 API。如何使用 PySpark 做到这一点？批处理的原因是因为 API 可能不会接受来自大数据系统的大量数据。

我首先想到的是LIMIT，但这不是“确定性的”。此外，它似乎效率低下？

【问题讨论】：

【解决方案1】：

df.foreachPartition { ele =>
   ele.grouped(1000).foreach { chunk =>
   postToServer(chunk)
}

代码在scala中，你可以在python中检查。它将创建 1000 个批次。

【讨论】：

只是好奇，repartition 创建所需大小的分区并以类似的方式（不分组）处理批次更好还是更好？如果是，为什么？我可以说重新分区一般来说太贵了吗？我还发现结果不是数据框。我应该如何使它成为数据框？有stackoverflow.com/a/48111699/292291，但想知道我是否可以不显式设置toDF() 中的列名？
`repartition` 算法会进行完全洗牌，并使用均匀分布的数据创建新分区。它很昂贵，它为您的应用程序提供了并行性。
出于某种原因，现在我得到“'itertools.chain' 对象没有属性'分组'”知道为什么吗？似乎itertools.chain 确实没有分组功能。我正在使用 PySpark 顺便说一句

【解决方案2】：

使用foreachPartition，然后使用类似how to split an iterable in constant-size chunks 的方法将可迭代对象批处理到 1000 个组中，可以说是最有效的 Spark 资源使用方式。

def handle_iterator(it):
    # batch the iterable and call API
    pass
df.foreachPartition(handle_iterator)

注意：这会从 executor 对 API 进行并行调用，并且在实践中可能不是这样，例如速率限制是个问题。

【讨论】：

spark中没有办法分组吗？另外，如果我在 Python 中进行分块，是否有任何理由使用 foreachPartition 而不是 map？
Map 是惰性的，不应包含任何副作用。它还一次访问单行，因此不适合批处理。如果你想在 spark 中批处理，有一个名为 collect_list 的聚合函数。但是，您需要找出产生甚至 1k 批次的分组/窗口。例如，对于提到的 10^8 行，您可以按哈希模 10^5 进行分组，这需要首先计算 df 大小，然后几乎可以肯定地对数据进行洗牌。