第 7 章 Trigger(触发器)
流式查询的触发器定义了流式数据处理的时间, 流式查询根据触发器的不同, 可以是根据固定的批处理间隔进行微批处理查询, 也可以是连续的查询.
| Trigger Type | Description |
|---|---|
| unspecified (default) | 没有显示的设定触发器, 表示使用 micro-batch mode, 尽可能块的处理每个批次的数据. 如果无数据可用, 则处于阻塞状态, 等待数据流入 |
| Fixed interval micro-batches 固定周期的微批处理 | 查询会在微批处理模式下执行, 其中微批处理将以用户指定的间隔执行. 1. 如果以前的微批处理在间隔内完成, 则引擎会等待间隔结束, 然后开启下一个微批次 2. 如果前一个微批处理在一个间隔内没有完成(即错过了间隔边界), 则下个微批处理会在上一个完成之后立即启动(不会等待下一个间隔边界) 3. 如果没有新数据可用, 则不会启动微批次. 适用于流式数据的批处理作业 |
| One-time micro-batch 一次性微批次 | 查询将在所有可用数据上执行一次微批次处理, 然后自行停止. 如果你希望定期启动集群, 然后处理集群关闭期间产生的数据, 然后再关闭集群. 这种情况下很有用. 它可以显著的降低成本. 一般用于非实时的数据分析 |
| Continuous with fixed checkpoint interval (experimental 2.3 引入) 连续处理 | 以超低延迟处理数据 |
// 1. 默认触发器
val query: StreamingQuery = df.writeStream
.outputMode("append")
.format("console")
.start()
// 2. 微批处理模式
val query: StreamingQuery = df.writeStream
.outputMode("append")
.format("console")
.trigger(Trigger.ProcessingTime("2 seconds"))
.start
// 3. 只处理一次. 处理完毕之后会自动退出
val query: StreamingQuery = df.writeStream
.outputMode("append")
.format("console")
.trigger(Trigger.Once())
.start()
// 4. 持续处理
val query: StreamingQuery = df.writeStream
.outputMode("append")
.format("console")
.trigger(Trigger.Continuous("1 seconds"))
.start