Flink WebUI解析（待更新）

写在前面

2024-05-23：对于flink webUI的查看，首先还是根据先整体观看一下Task的划分，然后根据Exception去寻找对应的TaskManager相关信息，然后双向确定是什么原因造成的；作为metrics，虽然看起来花里胡哨，本质上对flink底层实现机制的基本逻辑弄懂，应该还是比较容易看的；
注：由大佬有这块比较好的分享，跪求评论区共享一下；

2024-10-06：本次更新对Flink WebUI进行了全面介绍，剩下的待处理的监控指标包括：Flink内存模型，Java dump，【watermark，Accumulators，metrics，火焰图，】后续随缘补充一下

1. 代码示例

job代码：https://github.com/interestingcom/FlinkTest/blob/master/src/main/java/com/example/job/WordCount.java
以WordCount为例，说明Flink Webui界面参数解析，详细代码参考：
核心代码如下：

StreamExecutionEnvironment streamingEnvironment = FlinkEnvironmentFactory.createStreamingEnvironment();
streamingEnvironment.setParallelism(4);

SingleOutputStreamOperator<String> word = streamingEnvironment.addSource(new WordCountSource()).flatMap(new SplitMapFunction());

SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = word.map(value -> Tuple2.of(value, 1)).returns(Types.TUPLE(Types.STRING,Types.INT));

KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = wordAndOne.keyBy(value -> value.f0);

SingleOutputStreamOperator<Tuple2<String, Integer>> sum = tuple2StringKeyedStream.sum(1);

sum.print("flink计算结果===>");

streamingEnvironment.execute();

产生异常的代码：

if ( string.equals("world") && randomInt %5==0){
    throw new Exception("随机数是randomInt"+randomInt+"遇到单词world...");
}

2. WebUI详细信息

2.1. Submit New Job

位置1：上传jar包
位置2：传入main class；

2.2.Overview

位置1，黑色表示总共的subTask个数，绿色表示正在运行的subTask个数；

2.3. Runing Jobs

2.3.1 Overview

位置1：当前Job的概述，包括JobID，开始时间，持续时间
位置2：当前Job的监控信息，包括异常信息，时间线，快照信息，配置信息
位置3：当前Job的JobGraph（合并算子链后的Graph）
位置4：当前Job对应Task粒度的监控信息；Record received/sent表征数据在两个算子之间的传输条数（准确来说是已经处理的数据条数），Bytes received/sent表示数据传输的字节数（这里包括Barrier导致的字节数）

点击Source:Kafka2Redis的这个Task，位置1包括如下信息：
1、Details：从Task角度，统计输入/数据数据量，开始/持续/结束时间，Task状态，Task的实例数（并行度）
2、SubTasks：从SubTask角度统计Details；SubTasks可以理解为并行化的Task，即实际运行在各个节点上的Task，Task的实例化；More可以跳转到当前SubTask所在的TaskManager的日志信息
3、TaskManagers：从TaskManager角度统计Details信息，
4、Watermarks：埋点，暂未用到
5、Accumulators：埋点，暂未用到
6、BackPress：从SubTask角度给出，Backpressured / Idle / Busy占比，官网给出介绍如下：

• backPressuredTimeMsPerSecond：The time (in milliseconds) this task is back pressured (soft or hard) per second. It’s a sum of softBackPressuredTimeMsPerSecond and hardBackPressuredTimeMsPerSecond.
  注：软背压期间任务应该还能继续处理数据，硬背压任务被blocked
• idleTimeMsPerSecond：The time (in milliseconds) this task is idle (has no data to process) per second. Idle time excludes back pressured time, so if the task is back pressured it is not idle.
• busyTimeMsPerSecond：The time (in milliseconds) this task is busy (neither idle nor back pressured) per second. Can be NaN, if the value could not be calculated.

参考链接：Metrics和监控反压

7、Metrics：从SubTask角度查看监控指标，埋点，暂未用到
位置1：举例说明Task0在消费Kafka的时候，请求总数

8、Flame Graph：火焰图，用于描述Function在单位时间内被调用时长，宽度约宽，表示占用资源越多
位置1：用户自定义了map算子，命名伟UserMapFunction，其中12和47表示map函数在代码中被调用的位置
位置2：用户自定义函数Function2，持续在控制台打印6s的print语句，用于用户测试火焰图的信息，
位置3：用户自定义函数Function1，持续在控制台打印60s的print语句，用于对比用户2

一条数据经过map算子，if data = ‘a’ then Function1 else Function2，实际测试发现，我们尝试输入大量的数据!=a，则Function2的宽度明细高于Function1，所以可以看出火焰图只能反应函数被调用的执行时长，真正遇到问题只能做参考，还结合具体情况分析；

2.3.2.Exceptions

Root Exception：抛出Task异常，代码中插入throw new Exception("不允许用户传入key为error的数据");；平时在检索的时候注意观察cause by，Exception，error等关键字
Exception History：记录Exception的历史信息；
注：Exception信息不一定在TaskManager中会打印，此时去查看Timeline，会发现有个SubTask的时间线在Runing和failed切换（应该是这个SubTask不断重启导致的）

2.3.3.TimeLine

TimeLine显示了Flink任务在各个阶段的占比，Vertex Timeline从Task角度统计，SubTask TimeLine显示了各个Task实例的运行阶段占比，由于上面在Map算子中抛出了Exception，下面的SubTask0消费到error数据，因此会不断抛出异常
注：这里我们可以看到其他的SubTask没有标记为Failed，具体作业调度参考：作业调度

2.3.4.Checkpoint

位置1：统计Checkpoint此时，包括触发/正在处理/完成/失败/恢复次数；
位置2：最近一次Checkpoint的信息；End to End Duration：Checkpoint从触发到最后一次确认的时间，通常大于某一个Task的End to End Duration的时间；Checkpointed Data Size：Checkpoint大小；
位置3：Checkpoint详细信息，包括存储位置，是否对齐，丢弃等信息；
位置4：从Task角度（Operators，算子角度）和SubTask（Task实例）对Checkpoint信息进行描述
Operators角度：

• Acknowledged：已经完成的acknowledged（确认）的SubTask/总的Task数量；
• Latest Acknowledgment：JobManager 接收到任何 subtask 的最新确认的时间（如果尚未收到确认，则不适用）。
• Processed (persisted) in-flight data：Checkpoint对齐期间，处理/持久化字节数。

SubTask角度，同样的给出Minimum/Average/Maximum占比，以及不同SubTask具体Checkpoint情况

• Sync Duration：Checkpoint同步时长，包括Operator的快照状态，阻塞Subtask的其他行为；
• Async Duration：Checkpoint异步持续时间，包括写入文件系统所需时间，对于非对齐的Checkpoint统计时长参考官网
• Alignment Duration：处理第一个barrier和最后一个barrier之间的时间，提前接收到barrier的Channel会阻塞并停止处理后续数据，一般Forward的传递方式下，这里应该是0；
• Start Delay：从Checkpoint barrier创建到收到第一个Checkpoint barrier所用的时间；
• Unaligned Checkpoint：Checkpoint 完成的时候是否是一个 unaligned checkpoint。在 alignment 超时的时候 aligned checkpoint 可以自动切换成 unaligned checkpoint。

参考链接：监控 Checkpoint
位置6：History是对历史Checkpoint的统计，默认显示15次，Status：In Progress/Completed/Failed/Savepoint；
Summary从Minimum/Average/Maximum统计Checkpoint相关信息；Configuration给出的是关于Checkpoint相关的配置信息；

2.3.5.Configuration

Job的相关配置，可以看出位置1的相关信息说明了Task的执行模式，重启策略等信息；位置2是用户自定义的配置；

2.4 Task Managers

2.4.1 *TaskManagers Overview

列出所有的TaskManager
1、Path，ID：给出的是TaskManagerID和Akka通信地址
2、Data Port：数据传输端口，因为这里是createLocalEnvironmentWithWebUI，端口是-1，集群中从配置文件taskmanager.data.port读取
3、Last Heartbeat：上一次心跳时间
4、All/Free Slots：所有/空闲的Slot数
5、CPU cores：物理机的cpu核数，虚拟处理器=8
6、Physical MEM：物理机的内存数
7、JVM Heap Size：JVM堆内存大小，堆内存可以简单理解为用于存储对象
8、Flink Managed MEM：【后续补充内存模型，暂时可以不用关心，大部分情况没有用到】

2.4.2.Metrics

关于一些内存信息的坑，埋点，后续补充

2.4.3.Logs和Stdout

对于task Manager来说，核心主要关注日志文件；位置1是相关的日志信息和控制台打印输出，Log List给出日志列表
注：这里打印信息和任务运行环境，用户配置等相关，比如使用WebUI环境下，没有给出stdout的配置，只给出了Log的配置方式，生产中如果遇到异常信息，这两部分最好都看看；

2.4.4.Thread Dump

埋点，后续补充

2.5 Job Manager

JobManager的相关监控信息，Metrics涉及到Flink内存模型，同TaskManager；Configuration显示关于JobManager的配置信息，包括rest地址，TaskManager的内存，核数等配置信息（这里的配置更多描述的是集群的配置信息，前面的Job那块的Configuration描述的是Job相关的配置）；Logs/Stdout/Log List相关描述同TaskManager的描述；