如何在 Spark Structured Streaming 中将流式 DataFrame 写入多个接收器

Question

我有一组 SQL 规则，我需要将其应用于 foreachBatch() 内的流式数据帧。应用这些规则后，应将生成/过滤的数据帧写入多个目的地，如“delta”和“cosmos DB”。

以下是我尝试过的： 使用来自forEachBatch() 方法的静态数据框，我正在尝试创建一个临时视图，如下所示。

df.writeStream
  .format("delta")
  .foreachBatch(writeToDelta _)
  .outputMode("update")
  .start()

def upsertToDelta(microBatchOutputDF: DataFrame, batchId: Long) {
    microBatchOutputDF.createOrReplaceTempView("testTable")
}

但在运行代码时，它显示为表格或视图“testTable”未找到。

是否可以在 spark 结构化流中使用静态数据帧创建临时表/视图？

或者如何写入多个接收器？

Answer 1

来自 cmets 澄清 OPs 问题：

“我有一组 SQL 规则，我需要在 forEachBatch() 中的数据帧上应用这些规则。应用规则后，生成/过滤的数据帧将被写入多个目的地，例如 delta 和 cosmos DB。”

foreachBatch 允许您

• 重用现有的批处理数据源
• 写入多个位置

在您的情况下，我了解您希望在流数据帧上应用不同的转换并将其写入多个位置。你可以这样做：

df.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>

  // persist dataframe in case you are reusing it multiple times
  batchDF.persist()

  // apply SQL logic using `selectExpr` or just the DataFrame API
  val deltaBatchDf = batchDF.selectExpr("") 
  val cosmosBatchDf = batchDF.selectExpr("") 

  // write to multiple sinks like you would do with batch DataFrames
  // add more locations if required
  deltaBatchDf.write.format("delta").options(...).save(...)
  cosmosBatchDf.write.format("cosmos").options(...).save(...)

  // free memory
  batchDF.unpersist()
}