| Flink不触发窗口计算又不报错的问题定位 | 您所在的位置：网站首页 › flink水位线怎么算 › Flink不触发窗口计算又不报错的问题定位 |
Flink不触发窗口计算又不报错的问题定位
|
刚接触Flink不久，照着之前的一个老项目重构。好不容易写完代码并解决完各种报错，结果发现即使不报错也无法运出自己想要的效果。调试发现能正常消费到Kafka的消息，但是却无法触发窗口计算。在网上翻到一篇博客问题定位：Flink水位线不触发问题 ,上面说是watermark有问题，通过Flink的管理控制台发现watermark没生成： 既然定位到是watermark的问题，就知道解决问题的方向了。通过对比重构前可以运行的旧代码，发现了端倪。 旧代码大致如下： assignTimestampsAndWatermarks()方法返回的是一个全新的对象SingleOutputStreamOperator，而不再是原来那个DataStream，之所以第一种写法可以，是因为SingleOutputStreamOperator是DataStream的子类，看起来引用类型没发生变化，但实际上返回的对象已经发生了改变。后面要用assignTimestampsAndWatermarks()方法返回的对象来创建一个临时视图而不是addSource()返回的那个： DataStream callInfoStream = environment .addSource(new FlinkKafkaConsumer( config.getInputTopic(), new CallInfoSchema(), kafkaConf)) .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks() { private static final long serialVersionUID = 1L; private long currentMaxTimestamp = 0; @Override public long extractTimestamp(CallInfo element, long previousElementTimestamp) { long timestamp = element.getEndTime(); currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp); return timestamp; } @Override public Watermark getCurrentWatermark() { return new Watermark(currentMaxTimestamp); } }).setParallelism(1); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(environment); tableEnv.createTemporaryView( "CallInfo", callInfoStream, "traceType, traceId, elapsedTime, deviceId, callResult, streamId, streamType, " + "userId, groupId, childNum, recordType, startTime, endTime, dateStr, " + "successNum, failNum, rowtime.rowtime");搞了半天，原来是我代码写错了。这个问题坑就坑在它不会报错，也没有任何提示信息。 |
注意，截图的时候我的问题已经解决了，没有watermark的时候显示的是no watermark，而不是上图中的0
我重构后的错误代码： 
【本文地址】
公司简介
联系我们