关于30大洋看的一篇帖子(为什么我的Flink任务正常运行,UI上却不显示接收和发送的数据条数呢?)
Posted gxgd
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了关于30大洋看的一篇帖子(为什么我的Flink任务正常运行,UI上却不显示接收和发送的数据条数呢?)相关的知识,希望对你有一定的参考价值。
最近发现有好几个同学问我这个问题,为什么我的Flink任务正常运行,数据也可以打印,而且都保存到数据库了,但是UI上面却不显示数据接收和发送的条数,我都快被问疯了,今天就给大家详细说一下这个小问题.
首先先来复现一下这个问题,我们先看下面的代码(只是一部分代码)
def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.setParallelism(1) val ds = CommonUtils.getDataStream(env = env) .name("kafka-source") .filter(_.nonEmpty) .print() env.execute() }
代码非常的简单,没有任何的逻辑,从kafka读取数据,只是做了一个filter,然后直接print,我这里就不写sink,但是效果和直接sink是一样的,把这个任务提交到集群运行,效果如下图所示:
我已经向kafka写入数据了,这个地方怎么不显示呢?然后我们来看下tm的stdout,因为我代码里面直接print了,看下面的图
很明显数据打印了,说明我们程序是没有问题的,那问题在哪呢?其实并不是你的程序有问题,也不是Flink的UI有bug,是因为默认情况下Flink开启了operator chain,所以source filter print chain在了起就是在一个DAG里面,没有向下游发送数据,所以显示都为0,关于operator chain前面的文章已经说过了,还不了解的可以去查一下.那怎么能让他显示呢?
第一种方法,就是从metric里面看,可以自己添加两个metric,如下图所示
我添加了两个metric,一个filter的输入和filter的输出,可以看到都是100条数据.那如果想不添加metric,在ui上就能显示呢?这个时候就需要打断operator chain,具体有三种写法,如下代码所示
def main(args: Array[String]): Unit = { val env = StreamExecutionEnvironment.getExecutionEnvironment env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) env.setParallelism(1) val ds = CommonUtils.getDataStream(env = env) .name("kafka-source") .filter(_.nonEmpty) .startNewChain() .disableChaining() .setParallelism(2) .print() env.execute() }
这三种写法都可以达到打断chain的目的,有什么区别呢?startNewChain和disableChaining没有实质性的区别,他俩都会打断chain,但是不会改变算子的并发度,setParallelism和前面的算子并发度,设置的不一致自然就打断chain了.我们就演示一下第一个
可以看到上面的DAG图显示了两个,并且下面可以看到接收和发送的数据了,剩下的2种方法同样可以达到这样的效果,大家可以尝试一下.
Flink的operator chain是有利于提高程序的性能的,建议使用,如果想要打断operator chain又不想改变并发,就用前两种方法,如果想要改变并发就用第三张方法
30大洋 我尽力了
以上是关于关于30大洋看的一篇帖子(为什么我的Flink任务正常运行,UI上却不显示接收和发送的数据条数呢?)的主要内容,如果未能解决你的问题,请参考以下文章