印刷厂网站建设方案,企业门户网站 php,wordpress左边菜单,建设网站的价格是多少钱flink实时流遇到的问题排查1、技术和环境2、问题表述3、简化的代码4、问题排查思路5、结论6、后续补充1、技术和环境
技术#xff1a;kafka、zookeeper、DataStream、redis 环境表述#xff1a;kafka生产者KafkaProducerTest类mock 3条日志后#xff0c;FlinkDataRealTimeF…
flink实时流遇到的问题排查1、技术和环境2、问题表述3、简化的代码4、问题排查思路5、结论6、后续补充1、技术和环境
技术kafka、zookeeper、DataStream、redis 环境表述kafka生产者KafkaProducerTest类mock 3条日志后FlinkDataRealTimeFlowDeal类有建消费者消费日志数据进行实时流DataStream处理进行日志清洗、数据落库redis。
2、问题表述
理论上 KafkaProducerTest生产者每次执行x条日志消费者实际读取x条日志实际落库x条处理结果。
实际 有问题KafkaProducerTest生产者第1次执行3条日志消费者实际读取3条日志实际落库2条处理结果。 正常KafkaProducerTest生产者第2次执行之前的3条日志消费者实际读取3条日志实际落库3条处理结果。 有问题KafkaProducerTest生产者第3次执行之前的3条日志消费者实际读取3条日志实际落库2条处理结果。 正常KafkaProducerTest生产者第4次执行之前的3条日志消费者实际读取3条日志实际落库3条处理结果。 … 问题总结奇数次执行时数据漏掉了1条没有落库偶数次全部落库成功。 注意每次执行的是相同的日志数据测试用。
3、简化的代码
DataStream.flatMap(进行日志清洗-Collector收集).keyBy(0).countWindow(2).reduce(进行聚合).process(进行redis落库)4、问题排查思路
首先查看日志发现 进行【日志清洗-Collector收集】的方法内部的错误日志信息打印处理了3次符合要求说明3条日志均进行了日志清洗。
然后结合看方法flatMap(String value, CollectorTuple2String, List outflatMap扁平化得到的Collector是一个Tuple2String, List类型的收集。
接着keyby(0)后以Tuple2的第一个参数对数据进行平铺得到 String key1List UserEventAction111
String key2List UserEventAction11 String key2List UserEventAction22
String key3List UserEventAction1 String key3List UserEventAction2 String key3List UserEventAction3
第一条日志未处理key1平铺结果处理0条未处理1条 第二条日志处理key2平铺结果处理2条未处理0条 第三条日志部分处理key3平铺结果处理2条未处理1条 备注本业务场景一条日志对应一个key。
进行【聚合】的方法reduce内部共处理了4条Tuple2String, List UserEventAction里的平铺元素日志清洗的对象结果。缺少了2条未处理。
观察到对应日志条数未处理1条。处理了2条日志1条处理1条部分处理。
聚合前一步是countWindow。 推断是countWindow(2)出现问题。
上述结果推测 key1暂不处理具备1条数据在countWindow(2)时数据每满2个触发一次会处理。暂时不处理List UserEventAction111。等下一次key1有新数据时候满2处理。
key2处理具备2条数据在countWindow(2)时数据每满2个触发一次会处理2条List UserEventAction11List UserEventAction22。
key3部分处理具备3条数据在countWindow(2)时数据每满2个触发一次会处理2条List UserEventAction1List UserEventAction2。暂时不处理List UserEventAction3。等下一次key3有新数据时候满2条处理。
备注也可以从落库redis的数据反序列化后得到印证。
5、结论
把countWindow改成1就可以都落库了。
DataStream.flatMap(进行日志清洗-Collector收集).keyBy(0).countWindow(1).reduce(进行聚合).process(进行redis落库)问题解决了问题在countWindow(2)也就是当根据keyBy(0)分组之后数据的数量每次达到2时进行输出。
日志清洗后的结果是一条日志对应一个key一个key对应多个List UserEventAction或者单个List UserEventAction。
设置2的时候对应的key的List UserEventAction如果只有一条就不会落库kafka生产者执行两次时候就会累积到两条相同的key的数据每满2条处理后续操作所以之前有奇数次执行和偶数次执行区别。
6、后续补充
主要是和flatMap平铺后收集到的key的种类数量有关系。
