美团点评实时计算平台的Flink监控与告警实践 作者 | 孙梦瑶（美团点评） 整理 | 李培殿 & 杨伟海（Flink 社区志愿者） 校对 | 杨伟海（Flink 社区志愿者）本文由美团点评研发工程师孙梦瑶分享，主要介绍 Flink 的指标监控和报警，从以下四个方面展开： 监控告警链路：基于美团点评实时计算平台的实践 常用的监控项：哪些指标可以高效地衡量作业 指标的聚合方式：横看成岭侧成峰 指标监控的应用：有哪些常见的表达方式供参考 为什么我们关注指标监控 以天气为例。 指标：衡量和描述对象的方式 可量化：比如最近天气很热。今天比昨天热吗？北京的温度比上海更热吗？大家就没有办法评判，所以温度就是这样一个指标，来量化我们天热的程度。标准化：我们习惯说的温度是摄氏温度，如果有人跟你讲华氏温度，说今天 77 度，你就会觉得很奇怪，气温怎么会有这么高的数值，因此，我们的指标还需要是标准化的，需要有一个统一的标准。多维度：南方的同学觉得 35 度闷得喘不过气来；北方的同学觉得 35 度好像也就那样。因为我们除了气温这个指标会影响人体的舒适度之外，还有一个指标叫空气湿度。所以衡量天气需要结合多个维度的指标。 监控：对指标进行监测和控制 实时：比如天气预报，实时的预报才是我们需要的监控内容。易用：相比于电视机里固定时间播报的天气信息，手机 App 就是易用的天气监控软件。可查询历史：比如前几天某地一直在下雨，河流湍急，可能影响我出行的选择。 1. 监控报警的链路 1.1 监控报警链路 美团点评的指标监控报警的链路如下图所示。 首先是我们对日志和指标都会进行统一的集中化的收集。Reporter （2.8 和 3.1 中有介绍）把 Flink 作业的指标作为一条条日志打出来。然后再通过日志收集收上去，收到 Kafka 里面。接下来会通过实时作业做解析和聚合，再将得到的指标落到 Kafka 里，作为实时数据源。 下游会根据不同的需求，对不同的数据做不同的处理和展示。日志数据会落到 ES 里供查询使用，同时也会根据关键字接实时作业进行处理，做日志相关报警；数值指标会落到 OpenTSDB 里供大家查询，同时也支持各类的指标报警。最终这些内容还是会集中到我们的实时计算平台里，给用户做一个统一的展示。 整个链路下来，主要分为三个关键环节。 日志收集部分，我们首先是要把这些日志和指标进行统一化、集中化的收集。对于这一环，之前两个讲师也讲过， Flink 现在提供的方式有三种：一个是在 Flink UI 上可以直接看到这个作业的一些指标；第二种 REST API 从作业上获取指标；第三种就是配各种第三方的 Reporter 。美团这边是在 slf4j 的基础上增加自己的维度信息格式化后往下发。解析展示部分，使用一些 Flink 作业去解析聚合平台所有作业的指标数据，展示给用户，也提供给下游使用。监控和报警部分，对于聚合完成了的指标，做一些个性化的可配置的规则报警。 1.2 指标展示：Grafana Grafana 支持比较多的数据源格式，比如 ES、OpenTSDB 等；它有个变量的功能，可以看某个作业的指标，也可以一起对比看。 相比于自研的指标展示工具，Grafana 配置界面会比较方便，省时省力，性价比高。如果是只是想简单的展示一下所有的作业的指标的话，Grafana 是个很好的选择，它也可以被外嵌在其他的页面上。但是 Grafana 图的类型比较单一，在实际的直接使用过程中可能还会有一些局限性。 2. 常用的监控项 下面我们来关注下一般会使用哪些指标来衡量作业运行的状况。 2.1 常用的指标 系统指标 系统指标在 Flink 官网有相应的说明。 对于系统指标最常关注的是作业的可用性，如 uptime (作业持续运行的时间)、fullRestarts (作业重启的次数)。第二个关注的是作业的流量，可以通过 numRecordsIn、numBytesInLocal 等相关指标来关注作业每天处理的消息数目及高峰时间段的流量，通过关注这些指标可以观察作业的流量表现是否正常。然后是 CPU（如：CPU.Load）、内存（如：Heap.Used ）、GC ( 如：GarbageCollector.Count、GarbageCollector.Time ) 及网络 ( inputQueueLength、outputQueueLength) 相关指标，这些指标一般是用来排查作业的故障信息。另外是 checkpoint 相关信息，例如最近完成的 checkpoint 的时长（ lastCheckpointDuration ）、最近完成的 checkpoint 的大小（ lastCheckpointSize ）、作业失败后恢复的能力（ lastCheckpointRestoreTimestamp ）、成功和失败的 checkpoint 数目（ numberOfCompletedCheckpoints、numberOfFailedCheckpoints ）以及在 Exactly once 模式下 barrier 对齐时间（ checkpointAlignmentTime ）。还有就是 connector 的指标，例如常用的 Kafka connector ，Kafka 自身提供了一些指标，可以帮助我们了解到作业最新消费的消息的状况、作业是否有延迟等。自定义指标 自定义指标是指用户可以在自己的作业逻辑中进行埋点，这样可以对自己的业务逻辑进行监控。 正如其他讲师所提到的，现在的 Flink 作业更像是一种微服务，不仅关心作业是否把所有数据都处理完了，还希望作业可以 7×24 小时不间断的运行来处理数据。因此在业务逻辑中重要的指标在 Flink 中也会很重要。 比如处理逻辑耗时打点，例如包含复杂逻辑的业务系统，可以通过在逻辑前后进行打点，这样可以查看每条消息处理完这个逻辑的耗时。另一块是外部服务调用的性能， 在 Flink 作业中可能需要访问外部存储（如 Redis ）, 可以通过打点来查看请求的耗时、请求的成功率等。还有是缓存命中率，有时候由于数据集过大，我们只访问热数据，此时会在内存中缓存一部分信息，我们可以监控缓存命中率，如果缓存命中率非常高说明缓存有效，如果缓存命中率非常低，一直在访问外部存储，就需要考虑缓存设计的是否合理。另外还有几类是关于作业的处理逻辑，如果处理逻辑抛出异常将会导致作业 fullRestarts，此时一般会将这些异常进行 catch 住，如果涉及复杂计算的可以通过重试机制多试几次，如果重试后未成功则丢弃数据 。此时可以将处理数据的占比或者数据的某些特征作为指标上报，这样可以观察此类数据的占比来观测数据处理是否存在异常。又如 filter 过滤的数据占比可以观测 filter 的逻辑是否正常，还有窗口等涉及时间的算子需要监测超时丢弃的数据的占比等。 2.2 如何确定哪些指标需要关注？ 第一点是作业状态相关的， 如作业是否出故障、作业是否存活、作业是否稳定运行、影响作业可用性的风险因素（如上次 checkpoint 是否成功、最近成功的 checkpoint 的时间）。第二点是作业性能相关的，如作业的处理延迟、数据倾斜、性能瓶颈（如外部访问）等。第三点是业务逻辑相关，如上游数据质量、新上的逻辑是否存在问题、数据是否存在丢失（ Exactly once 语义中数据是否允许丢失）。 3. 指标的聚合方式 在上面介绍了常用的监控指标，接下来介绍下这些指标应该怎么看。同一个指标可能在机器的角度去看，也可能在作业的角度去看，不同的角度会得出不同的结果。 首先是作业的聚合维度，细粒度的如 Task、Operator 维度，稍微大点的粒度如 Job、机器、集群或者是业务维度（如每个区域）。当查问题时从大的粒度着手，向细粒度进行排查。如果想看全局的现状则需要比较粗的粒度。可以将原始指标进行上报然后根据不同场景进行聚合。如果要做性能测试则需要细粒度的查询，如 task 粒度。 另一方面是聚合的方式，如总和、均值、最大值、最小值、变化率等，需要注意是要消除统计误差，对数据取移动平均或者固定时间的平均来使曲线变得更加平滑。还有是差值，如上游数据量与下游数据量的差值、最新 offset 与消费的 offset 的差值。另外对于衡量 xx 率、xx 耗时可以使用 99 线。最后还有一点需要关注的，也是我们在实际工作中遇到的坑，即指标的缺失，如果没有拿到指标，作业状态则变成了黑盒，需要去关注作业的指标收集是否正常，需要监测是否存在指标丢失，是单个指标丢失还是整个作业的指标丢失。 最后是在观察指标的时候可能需要多个指标复杂聚合查询，如常见的时间线对比，例如之前正常的作业在今天出现反压，可以通过查询今天数据量的同比昨天数据量的增长。另外不同的业务有不同的趋势，例如外卖存在高峰时间段，可以对比数据量在高峰时间段的环比增长来进行衡量。还有关注的指标的持续时间，如作业的数据延迟，如果延迟时间较长则作业可能存在异常；还有指标的周期性，如果指标的变化存在周期性，则考虑是否因为时间窗口的影响。 还有需要考虑的是结合外部系统进行计算，例如上游为消费 Kafka ，除了想知道当前消费的状况，还想查看上游的数据量。例如该图中，蓝线为上游 Kafka 的数据量，红线为作业的 source 算子的 output 数据量，可以看到在午高峰和晚高峰基本上是持平的状态， 上游数据在午高峰及晚高峰有较高的增长，虽然在高峰时刻有反压，但主要原因是由于上游数据量的增长而不是由于作业的处理能力不足。如果上游有多个算子可以将多个算子的数据量进行相加，这也是我们除了使用 Grafana 外还自研的前端进行展示的原因，自研前端可以将指标更加灵活的进行展示。 4. 指标监控的应用 4.1 作业异常报警 作业状态异常：包括作业任务的异常状态如 failing，也包括 uptime 等指标的异常。作业无指标上报：作业无指标上报会给作业的负责人发报警；当上报的作业多到一定程度了，达到预值的时候会直接给平台的管理员发报警。指标达到阈值：是大家最常用的报警项。比如：处理量跌 0消费延迟（落后一定数量、持续一定时间）失败率、丢失率等个性化：实时计算平台中有很多类任务，不同的任务它会有不同的特性。比如：报警时段：不同的时间段报警，可能需要有不同的域值，或者不同的报警方式（公司通讯软件报警、电话报警等）聚合方式：不同的业务可能会有不同的报警的聚合的方式，这个也是需要尽量兼容的。错误日志、关键词日志：当错误日志到达一定量或者日志出现某关键词时，触发报警。注意：报警系统本身的稳定性，放到第 1 位，避免出现误报、漏报、延迟。否则会影响业务方的准确判断。 4.2 指标大盘 反映平台整体的现状： 异常值高亮 多维度聚合 时间线对比等 及时发现并快速定位到故障给出平台可优化的方向便于统筹资源分配 4.3 自动化运维 运维的几种阶段： 无法运维：没有指标，作业状态是个黑盒，出了问题一群人查代码。手动运维：重启，扩容，回滚、迁移，降级，纠正错误代码，优化处理逻辑。手动运维表示无论在干什么，当报警电话一来，你需要掏出电脑、手机去排查问题。辅助运维：当手动运维做多了，把大家的业务作业的各项指标都进行标准化，我们就可以得到一些参考值。把这些经验汇总，作为其他同学的运维的时候参考的建议，这样即使是新人也可以快速借助这些辅助工具进行处理，降低学习成本。智能运维：智能运维是不需要人处理，当发生故障的时候，自动操作的运维方式。执行作业的机器挂了，自动拉起，自动把作业启动起来。资源不足了，自动去扩容。线上的作业有问题，自动切换到备用的作业……当然目前能做到的这些只能解决一部分问题，一些代码问题带来的故障还是需要人为介入修复 bug。 Q&A Q1：构建一整套指标系统，指标库如何维护？需要去对程序进行代码级别的修改，还是修改配置即可？ A：既然想做一整套的监控系统自然希望这个指标尽可能是一个可适配的方式，那么我们需要做什么？ 在设计整套系统的架构时，需要有一定的兼容性，不能只关注一类指标。 设计初期需要考虑有哪些类型的指标，每个类型的指标有什么样的特征，可能有哪些聚合的维度，用什么样的方式去聚合。 搭建模型。 设计，先把指标的特征提取出来，然后对这些特征去进行设计，最后做一个能兼容的系统，这样对于已知类型的指标，就只需修改配置就可以扩展了。 Q2：Grafana 平台的展示效果很好，但是报警不友好；报警这块有比较成熟的工具吗？ A：可以看看 Prometheus，报警还是挺成熟的。报警比指标聚合更需要个性化的东西，如果需要功能非常完善的话，可能都需要考虑自研。 Q3：算子内部可以获取到 taskManager 的指标吗？ A：通过 restful API 去拿，不推荐在算子内部做，指标这个东西本身不应该影响你作业本身的处理逻辑，监控应该是一个比较外围的东西。 Q4: 如何根据指标发现作业问题的根源？ A: 按照指标从粗到细，可以参考 2.8 节和 3.1 节老师的教程。 Q5: 指标数据量比较大，如何选择存储？ A: 可以选择 openTSDB，其他 TSDB 也是可以的，像其他 Hive 或者 OLAP 引擎 也是可以考虑的，指标数据作为一种时序数据，目前已有很多成熟的方案可以选择。 视频回顾： https://ververica.cn/developers/flink-training-course-operation/ 推荐阅读： 对 Apache Flink 感兴趣的开发者，可以点击「阅读原文」查看“Apache Flink 零基础入门到进阶”（https://www.infoq.cn/theme/28） 系列文章，全面了解 Apache Flink 的技术细节，更加得心应手地使用 Apache Flink。 你也「在看」吗？?? 阅读原文