Trigger.Once 与 Azure 事件中心答案

【问题标题】：Trigger.Once with Azure Event HubsTrigger.Once 与 Azure 事件中心
【发布时间】：2020-06-16 16:34:42
【问题描述】：

我正在阅读有关将 Streaming 与 Trigger.Once 结合使用而不是批处理 (https://databricks.com/blog/2017/05/22/running-streaming-jobs-day-10x-cost-savings.html) 的信息。我使用 Azure 事件中心作为源，每个触发器有 10.000 个项目限制，我必须处理几百万个事件。这将导致仅处理 10.000 个项目，并且当您启用 Trigger.Once 时流将停止。有没有人知道如何使用流和触发器来做到这一点。一旦类似功能并在流为空或处理到设定时间时停止？

【问题讨论】：

我也有同样的问题！

标签： spark-structured-streaming azure-eventhub

【解决方案1】：

最近有机会实现类似的东西：

我们的需要是；

Databricks 将启动作业并从 eventthub 的上次运行偏移量读取一次
将数据保存到位置
结束作业，集群返回空闲/终止模式

这就是我们实现它的方式：

设置 eventthub conf：

val eventHubsConf: EventHubsConf = EventHubsConf(connStr).setMaxEventsPerTrigger(<someNumber>).setStartingPosition(EventPosition.fromOffset("@latest"))

从 eventthub 读取：

val incomingStream = spark.readStream.format("<your_evenhub_name>").options(eventHubsConf.toMap).load()

使用 Trigger.Once 编写

incomingStream.writeStream
  .format("parquet").outputMode("append").trigger(Trigger.Once())
  .partitionBy("<yourpartitionColumn>").option("truncate", false)
  .option("checkpointLocation", "<yourcheckpointlocation>")
  .option("path", "<youroutputdataDir>")
  .start()

【讨论】：