无流时触发结构化流窗口

Question

我想记录从火花结构化流的传入流读取到数据库的记录数。我正在使用ForeachBatch来转换传入的流批处理和写入所需的位置。我想日志0记录读取，如果在一个特定的小时没有记录。但是当没有流时，Foreach批处理不会执行。有人能帮我吗？我的代码如下：

val incomingStream=spark。读流。格式（“事件中心”）。选项（customEventhubParameters.toMap）。load（）

val query=incomingStream.writeStream.foreachBatch{
  (batchDF: DataFrame, batchId: Long)=> writeStreamToDataLake(batchDF,batchId,partitionColumn,fileLocation,errorFilePath,eventHubName,configMeta)
}
              .option("checkpointLocation",fileLocation+checkpointFolder+"/"+eventHubName)
              .trigger(Trigger.ProcessingTime(triggerTime.toLong))
              .start().awaitTermination()

Answer 1

这就是它的工作原理，甚至只有当有需要处理的内容，从而改变流的状态时，才会调用StreamingQueryListener的mods扩展。

可能还有另一种方法，但我会说"跳出框框思考"和pre-popualte与0每个时间帧这样的数据库，当查询AGGRegate，你会有正确的答案。

https://medium.com/@johankok/structured-streaming-in-a-flash-576cdb17bbee可以提供一些见解，以及火花：最终指南。