apache-spark - 如何在pyspark中使用foreach sink？

Question

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

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

def func():
    ops(wordCounts)

Answer 1

Spark 2.4.0 中添加了对 Python 中 foreach sink 的支持，并更新了文档:http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch

确保您拥有该版本并且您现在可以执行以下操作:

def process_row(row):
    # Process row
    pass

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