Flink 实现指定时长或消息条数的触发器
Flink 中窗口是很重要的一个功能,而窗口又经常配合触发器一起使用。
Flink 自带的触发器大概有:
CountTrigger: 指定条数触发 ContinuousEventTimeTrigger:指定事件时间触发
ContinuousProcessingTimeTrigger:指定处理时间触发 ProcessingTimeTrigger: 默认触发器,窗口结束触发
EventTimeTrigger: 默认处理时间触发器,窗口结束触发 NeverTrigger:全局窗口触发器,不触发
但是没有可以指定时间和条数一起作为触发条件的触发器,所有就自己实现了一个(参考:ProcessingTimeTrigger、CountTrigger)
看下调用触发器的窗口代码:
val stream = env.addSource(kafkaSource)
.map(s => {
s
})
.windowAll(TumblingProcessingTimeWindows.of(Time.seconds(60)))
.trigger(CountAndTimeTrigger.of(10, Time.seconds(10)))
.process(new ProcessAllWindowFunction[String, String, TimeWindow] {
override def process(context: Context, elements: Iterable[String], out: Collector[String]): Unit = {
var count = 0 elements.iterator.foreach(s => {
count += 1
})
logger.info("this trigger have : {} item", count)
}
})
很简单的一段代码:定义了一个60秒的窗口,触发器是自己实现的10条数据或者 10 秒触发一次的触发器,窗口函数就输出窗口数据的条数
下面看下自定义触发器 CountAndTimeTrigger 的核心代码如下:
/**
* CountAndTimeTrigger : 满足一定条数和时间触发
* 条数的触发使用计数器计数
* 时间的触发,使用 flink 的 timerServer,注册触发器触发
*
* @param <W>
*/
public class CountAndTimeTrigger<W extends Window> extends Trigger<Object, W> {
private Logger logger = LoggerFactory.getLogger(this.getClass());
// 触发的条数
private final long size;
// 触发的时长
private final long interval;
private static final long serialVersionUID = 1L;
// 条数计数器
private final ReducingStateDescriptor<Long> countStateDesc =
new ReducingStateDescriptor<>("count", new ReduceSum(), LongSerializer.INSTANCE);
// 时间计数器,保存下一次触发的时间
private final ReducingStateDescriptor<Long> timeStateDesc =
new ReducingStateDescriptor<>("fire-interval", new ReduceMin(), LongSerializer.INSTANCE); public CountAndTimeTrigger(long size, long interval) {
this.size = size;
this.interval = interval;
} @Override
public TriggerResult onElement(Object element, long timestamp, W window, TriggerContext ctx) throws Exception {
// 注册窗口结束的触发器, 不需要会自动触发
// ctx.registerProcessingTimeTimer(window.maxTimestamp());
// count
ReducingState<Long> count = ctx.getPartitionedState(countStateDesc);
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc);
// 每条数据 counter + 1
count.add(1L);
if (count.get() >= size) {
logger.info("countTrigger triggered, count : {}", count.get());
// 满足条数的触发条件,先清 0 条数计数器
count.clear();
// 满足条数时也需要清除时间的触发器,如果不是创建结束的触发器
if (fireTimestamp.get() != window.maxTimestamp()) {
// logger.info("delete trigger : {}, {}", sdf.format(fireTimestamp.get()), fireTimestamp.get());
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
}
fireTimestamp.clear();
// fire 触发计算
return TriggerResult.FIRE;
} // 触发之后,下一条数据进来才设置时间计数器注册下一次触发的时间
timestamp = ctx.getCurrentProcessingTime();
if (fireTimestamp.get() == null) {
// long start = timestamp - (timestamp % interval);
long nextFireTimestamp = timestamp + interval;
// logger.info("register trigger : {}, {}", sdf.format(nextFireTimestamp), nextFireTimestamp);
ctx.registerProcessingTimeTimer(nextFireTimestamp);
fireTimestamp.add(nextFireTimestamp);
}
return TriggerResult.CONTINUE;
} @Override
public TriggerResult onProcessingTime(long time, W window, TriggerContext ctx) throws Exception { // count
ReducingState<Long> count = ctx.getPartitionedState(countStateDesc);
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeStateDesc); // time trigger and window end
if (time == window.maxTimestamp()) {
logger.info("window close : {}", time);
// 窗口结束,清0条数和时间的计数器
count.clear();
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
} else if (fireTimestamp.get() != null && fireTimestamp.get().equals(time)) {
logger.info("timeTrigger trigger, time : {}", time);
// 时间计数器触发,清0条数和时间计数器
count.clear();
fireTimestamp.clear();
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
}
主要是在数据进来的时候,调用 onElement 做条数的计数器,满足条件就触发, onProcessingTime 是 flink 的 timeservice 调用的,作为定时触发的触发器
在时间和条数的定时器都有清除时间和条数计数器的计数,让计数器在下一条数据到的时候,重新开始计数
特别需要注意:窗口结束的时候,会自动触发调用 onProcessingTime ,一定要包含在触发器逻辑里面,不然不能获取窗口的完整数据
// time trigger and window end
if (time == window.maxTimestamp()) {
logger.info("window close : {}", time);
// 窗口结束,清0条数和时间的计数器
count.clear();
ctx.deleteProcessingTimeTimer(fireTimestamp.get());
fireTimestamp.clear();
return TriggerResult.FIRE_AND_PURGE;
}
如在获取到窗口触发时间是窗口的结束时间(即窗口的结束时间减1,Java的时间精度是到毫秒,如 10秒的窗口时间是:(00000, 10000)0000-10000 ,实际上窗口结束时间就是 9999)
看执行的结果:
从 “14:42:00,002 INFO - window close : 1573281719999” 窗口结束
到 “14:42:10,015 INFO - countTrigger triggered, count : 10 ” , “14:42:19,063 INFO - countTrigger triggered, count : 10” 条数触发
到 “14:42:36,499 INFO - timeTrigger trigger, time : 1573281756496” 时间触发
最后 窗口结束 “14:43:00,002 INFO - window close : 1573281779999”
搞定
完整代码:https://github.com/springMoon/flink-rookie/tree/master/src/main/scala/com/venn/stream/api/trigger
欢迎关注Flink菜鸟公众号,会不定期更新Flink(开发技术)相关的推文
最新文章
- Nginx+PHP On windows
- 解决file_get_contents无法请求https连接的方法
- 字节流和字符流(InputStream类和OutputStream类)
- js中退出语句break,continue和return 比较 (转载)
- Android-NDK编译:cocos2d-x(三) eclipse 导入工程
- Centos中yum方式安装java
- QThread 与 QObject的关系(QObject可以用于多线程,可以发送信号调用存在于其他线程的slot函数,但GUI类不可重入)
- sublime text3 好用的插件!!!
- OBJ-C中dispatch_once的用法
- iOS Swift开发的一些坑
- CentOS 7系统初始化
- mybatis百科-列映射类ResultMapping
- Android ViewPager + Fragment实现滑动页面
- Hopcroft-Carp 算法模板 自用
- python学习-(__new__方法和单例模式)
- 【WebService】调用第三方提供的webService服务(七)
- CNN初探
- Redis的持久化数据
- 大数据时代快速SQL引擎-Impala
- 使用svgdeveloper 和 svg-edit 绘制svg地图