如何使用 KeyedProcessFunction 实现在 apache flink 中进行多个 window 聚合?
How can I do multiple window aggregations in apache flink using KeyedProcessFunction implementation?
我想扩展较低的 window 聚合来计算较高的 window 聚合。
我的下层 window 聚合正在使用 KeyedProcessFunction,并且实现了 onTimer 以便在 window 结束时将数据刷新到接收器中。
我的代码与此处解释的类似:https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/learn-flink/event_driven/。但我还需要计算更高 windows 的聚合,
Input -> window(15 secs) -> window(1 min) -> window(15 min) -> ...
... ->window(1 day) -> out_1 -> out_2 -> out_3 ... ->out_n
如何扩展较低的 window 结果以计算较高的 window 聚合?
您可以从较低层的窗口中获取输出并将其发送到较高层,例如:
val hourlyTips = fares
.map((f: TaxiFare) => (f.driverId, f.tip))
.keyBy(_._1)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.reduce((f1: (Long, Float), f2: (Long, Float)) => { (f1._1, f1._2 + f2._2) })
val dailyTips = hourlyTips
.keyBy(_._1)
.window(TumblingEventTimeWindows.of(Time.hours(24)))
.reduce((f1: (Long, Float), f2: (Long, Float)) => { (f1._1, f1._2 + f2._2) })
hourlyTips.addSink(...)
dailyTips.addSink(...)
无论窗口是使用 DataStream windows 实现的,还是 windows 您自己使用 KeyedProcessFunction 实现的,这种基本方法都应该有效。
我想扩展较低的 window 聚合来计算较高的 window 聚合。
我的下层 window 聚合正在使用 KeyedProcessFunction,并且实现了 onTimer 以便在 window 结束时将数据刷新到接收器中。
我的代码与此处解释的类似:https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/learn-flink/event_driven/。但我还需要计算更高 windows 的聚合,
Input -> window(15 secs) -> window(1 min) -> window(15 min) -> ...
... ->window(1 day) -> out_1 -> out_2 -> out_3 ... ->out_n
如何扩展较低的 window 结果以计算较高的 window 聚合?
您可以从较低层的窗口中获取输出并将其发送到较高层,例如:
val hourlyTips = fares
.map((f: TaxiFare) => (f.driverId, f.tip))
.keyBy(_._1)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.reduce((f1: (Long, Float), f2: (Long, Float)) => { (f1._1, f1._2 + f2._2) })
val dailyTips = hourlyTips
.keyBy(_._1)
.window(TumblingEventTimeWindows.of(Time.hours(24)))
.reduce((f1: (Long, Float), f2: (Long, Float)) => { (f1._1, f1._2 + f2._2) })
hourlyTips.addSink(...)
dailyTips.addSink(...)
无论窗口是使用 DataStream windows 实现的,还是 windows 您自己使用 KeyedProcessFunction 实现的,这种基本方法都应该有效。