结构化流写入多个流

Question

我的场景

从流中获取数据并调用 return 一个 json 字符串的 UDF。 JSON 字符串中的属性之一是 UniqueId，UDF 将其生成为 guid.newGuid() (C#)。
1. UDF 的 DataFrame 输出被写入多个 streams/sinks 基于一些过滤器。

问题：

每个接收器正在获取由 UDF 生成的 UniqueId 的新值。我如何为所有接收器保持相同的 UniqueId。
如果每个接收器获得不同的 UniqueId 值，是否意味着我的 UDF 被每个接收器多次调用？
如果 UDF 被调用两次，调用它一次然后将相同数据写入不同接收器的选项是什么

inData = spark.readstream().format("eventhub")

udfdata = indata.select(from_json(myudf("column"), schema)).as("result").select(result.*)

filter1 =  udfdata.filter("column =='filter1'")
filter 2 = udfdata.filter("column =='filter2'") 

# write filter1 to two differnt sinks
filter1.writestream().format(delta).start(table1)
filter1.writestream().format(eventhub).start()

# write filter2 to two differnt sinks
filter2.writestream().format(delta).start(table2)
filter2.writestream().format(eventhub).start()

Answer 1

每次调用 .writestream()....start() 您都在创建一个新的独立流式查询。

这意味着对于您定义的每个输出接收器，Spark 将从输入源再次读取并处理数据帧。

如果您只想读取和处理一次然后输出到多个接收器，您可以使用 foreachBatch 接收器作为解决方法：

inData = spark.readstream().format("eventhub")
udfdata = indata.select(from_json(myudf("column"), schema)).as("result").select(result.*)

udfdata.writeStream().foreachBatch(filter_and_output).start()

def filter_and_output(udfdata, batchId):
    # At this point udfdata is a batch dataframe, no more a streaming dataframe
    udfdata.cache()
    filter1 = udfdata.filter("column =='filter1'")
    filter2 = udfdata.filter("column =='filter2'") 

    # write filter1
    filter1.write().format(delta).save(table1)
    filter1.write().format(eventhub).save()

    # write filter2
    filter2.write().format(delta).save(table2)
    filter2.write().format(eventhub).save()

    udfdata.unpersist()

您可以在 Spark Structured Streaming documentation.

中了解有关 foreachBatch 的更多信息

回答您的问题

如果您使用 foreachBatch，您的数据将只被处理一次，并且所有接收器都将具有相同的 UniqueId
是
使用 foreachBatch 将解决问题

结构化流写入多个流

structured streaming writing to multiple streams

apache-spark

spark-structured-streaming

azure-databricks