Spark结构化流聚合，没有数据时间戳（基于触发器的聚合）

作者: trpnest
发布时间: 2024-12-24 09:58:37 (2天前)
转自：

3 条回复

0#
回复此人
妖邪 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> 您可以创建自己的接收器并在每个addBatch（）调用上执行操作： </p> <pre> <code> class CustomSink extends Sink { override def addBatch(batchId: Long, data: DataFrame): Unit = { data.groupBy().agg(sum("age") as "sumAge").foreach(v => println(s"RESULT=$v")) } } class CustomSinkProvider extends StreamSinkProvider with DataSourceRegister { def createSink( sqlContext: SQLContext, parameters: Map[String, String], partitionColumns: Seq[String], outputMode: OutputMode): Sink = { new PersonSink() } def shortName(): String = "person" } </code> </pre> <P> 将outputMode设置为Update并每隔X秒触发一次 </p> <pre> <code> val query = ds.writeStream .trigger(Trigger.ProcessingTime("1 seconds")) .outputMode(OutputMode.Update()) .format("exactlyonce.CustomSinkProvider") </code> </pre> </DIV>

编辑
1#
回复此人
star*위위 | 2019-08-31 10-32

<div class =“post-text”itemprop =“text”> <P> “按处理时间触发”是否符合您的要求？ “按处理时间触发”触发每个间隔（由代码定义）。 </p> <P> 示例触发器代码如下。 </p> <P> <a href =“https://github.com/apache/spark/blob/v2.2.0/sql/core/src/main/scala/org/apache/spark/sql/streaming/ProcessingTime.scala#L34"rel =“nofollow noreferrer”> https://github.com/apache/spark/blob/v2.2.0/sql/core/src/main/scala/org/apache/spark/sql/streaming/ProcessingTime.scala#L34 </A> </p> </DIV>

编辑

登录后才能参与评论