如何在 pyspark 中使用 foreach 接收器？答案

【问题标题】：How to use foreach sink in pyspark?如何在 pyspark 中使用 foreach 接收器？
【发布时间】：2018-06-20 11:06:48
【问题描述】：

如何在 Python Spark 结构化流中使用 foreach 来触发输出操作。

query = wordCounts\
    .writeStream\
    .outputMode('update')\
    .foreach(func)\
    .start()

def func():
    ops(wordCounts)

【问题讨论】：

【解决方案1】：

确保您拥有该版本并且您现在可以这样做：

def process_row(row):
    # Process row
    pass

query = streamingDF.writeStream.foreach(process_row).start()

【讨论】：

【解决方案2】：

现在使用任何简单的技巧都无法在pyspark 中使用foreach，此外，在pyspark 中，update 输出模式仅准备用于调试。

推荐你在scala使用spark，不难学。

【讨论】：

【解决方案3】：

您可以改用DataFrame.foreach(f)。

【讨论】：

【解决方案4】：

TL;DR 无法在 pyspark 中使用 foreach 方法。

引用 Spark Structured Streaming 的 official documentation（突出显示我的）：

foreach 操作允许对输出数据进行任意操作。从 Spark 2.1 开始，这仅适用于 Scala 和 Java。

【讨论】：

非常感谢，我想知道在结构化流中使用 foreach 是否有欺骗的方法？是否可以在结构化流中使用缓存或持久化？
介意详述“在结构化流中使用foreach有什么技巧吗？”？没有缓存或持久化，因为每个触发器的 DataFrame 是“非常临时”的数据结构，Spark“Streams”的目的是您不应该关心低级执行细节。
"foreachBatch(...) [...]。从 Spark 2.4 开始，Scala、Java 和 Python 都支持此功能。因此，从 Spark 2.4 开始支持 Python。