Flink Forward 2018 - 流计算平台的运维优化分享

Flink Forward 是由 Apache 官方授权，用于介绍 Flink 社区最新动态、发展计划以及各一线大厂围绕 Flink 生态的生产实践经验的会议。Flink Forward 以前只在美国和德国举办，2018年12月20日首次来到中国。腾讯云大数据团队参加了会议并在会上介绍团队在公有云流计算平台服务化过程中的一些监控运维经验。

背景介绍

Stream Compute Service（简称 SCS）是腾讯公有云上的一站可视化流计算平台，它是一个位于云端的流式数据汇聚以及计算的服务，目标客户群体主要是 外部客户 以及 有上云需求的内部客户；它包含两部分，分别是流汇聚和流计算，流汇聚是云端的数据管道，一个云端的结构化的分布式消息队列； 流计算则是基于 Apache Flink 的流计算引擎，当前主推的是 StreamSQL；如下图，SCS 具备的优势包含：高吞吐低延时、弹性、易用、全托管等等

实践证明 Flink 是一个非常优秀的流计算框架，不管是在高吞吐低延时、还是在状态管理、流语义等方面都有出色的表现，但是，我们在运维过程中也发现了一些稳定性相关的问题，比如 OOM，虽然 Flink 自身提供了丰富的 Metric 指标，但它在作业状态分析、监控以及在线优化方面还比较薄弱，SCS 需要一个相对完善的外部系统实时监控线上作业，及时识别潜在的异常并提前进行预警或者触发作业在线优化，为云上的业务保驾护航。SCS 的监控系统已经过了3个阶段的发展，不同阶段面对不同的挑战、解决不同的问题：

1. 简单的监控系统，解决了从无到有的问题；
2. 基于经验规则的智能监控系统，可以做到提前预警和自动在线优化；
3. 基于机器学习的智能监控系统，能够更好的解决异常检测及负载预测等问题。

基础监控系统

这是一个比较简单的事后监控告警系统，Flink 作业通过 PerJob 模式在 Yarn 上运行，支撑服务周期性检查 Yarn Application 以及 Flink Job 的状态，当发现异常时发送告警。显然只有作业异常了才发送告警，没有提前预警以及在线优化的能力，监控的不完善导致 SCS 在第一阶段遇到了不少挑战，部分总结如下：

1. 告警不及时：收到告警时作业已处于异常状态，SCS 需要预警能力，提前识别潜在的作业异常，在发现作业可能会出现问题时立刻发送告警或者自动进行在线优化；
2. 异常漏报：由于是周期性状态检查，所以存在作业中间状态漏报的可能，比如作业频繁 restarting，而 restart 的时间点在检测周期中间，这时就可能捕捉不到状态异常，产生异常漏报现象；
3. 作业不稳定：比如使用 HeapStateBackend 运行作业，在 Continuous join 或者 query 场景下比较容易出现 OOM，Flink 没有提供自适应的 StateBackend，在大状态场景下在线将内存中的状态迁移到 RocksDB。

针对这些问题，我们对监控系统及 Flink 内核进行优化，这就是第二阶段基于经验规则的监控系统。

基于经验规则的监控系统

指标分析

Flink 本身提供了各种各样的 Metric，有 CPU、内存、GC、Operator latency 等等，每个 Metric 都能从不同的角度表征作业内部的状态，所以第二阶段，我们将精力集中在对指标的分析上。首先，我们通过 KafkaMetricReporter 实时将线上作业各类 Metric 指标采集到 Kafka，接着通过预处理及分组后使用 Flink 的 CEP 定制各种各样的规则，一个规则负责一类指标的在线模式匹配及分析。

当通过规则匹配到潜在异常时监控系统输出相应的 Event，如果是普通 Event，仅发送告警，如果是特殊 Event，比如需要在线进行动态扩缩容或者需要进行在线状态迁移，监控系统不仅发送告警，也会回调 JobManager 的 rest 接口，触发在线优化。

Event 类型

根据事件的严重程度，SCS 把 Event 分成三个等级：INFO、WARN、FATAL；INFO 级别的事件，比如 CPU、内存使用率偏低，可以低优先级处理；WARN 级别的事件，比如 Source 或者 Sink 吞吐量快速下降，需尽快处理；FATAL 级别的事件需立刻处理，比如作业处于 RESTARTING 状态。根据事件所属的进程，将事件分为 JobManagerEvent 及 TaskManagerEvent，部分事件如下：

每个 Event 对应到一条规则，规则通过 CEP Pattern 表达，根据规则复杂度，Pattern 可以是 Single 类型，也可以是 Composite 类型，如下是 TASK_MANAGER_HEAP_MEM_LOAD_HIGH 的 Pattern 样例：

以下是内存监控的效果图示，当 TaskManager 内存达到设定的阈值时，监控系统便会发送告警，同时触发作业的在线状态迁移，无缝完成线上作业优化。

自适应状态管理

很大一部分流计算作业是有状态的，状态如 Kafka consume offset、窗口中间数据、用户自定义状态等；相比业界其他流计算框架如 Storm 和 Spark，Flink 在状态存储及恢复方面的设计更加完善；从存储方式角度划分，Flink 提供了两类实现，分别是 HeapStateBackend 和 RocksDBStateBackend，前者状态缓存于内存，因此能够提供高吞吐低延时状态读写，但受限于内存大小，无法缓存大量状态，后者将状态缓存于磁盘，能够缓存大量状态，但状态读写吞吐量较低，延时也高。SCS 尝试将两种实现结合起来，实现自适应的状态管理，优先使用 HeapStateBackend，当监控系统发现内存有瓶颈时，调用拓展的 JobManager rest api，在线切换到 RocksDBStateBackend，相反，当内存有盈余时再切换回 HeapStateBackend，充分发挥两种状态管理器的优势。

从用途上划分，StateBackend 有两个子类，分别是 OperatorStateBackend 和 KeyedStateBackend，OperatorStateBackend 主要存储 Task 级别轻量级状态，比如 Kafka consume offset，所以不是瓶颈，也不是我们优化的重点；而 KeyedStateBackend 是 key 级别的存储器，状态数量和 key 数量正相关，所以容易成为瓶颈，这是我们优化的重点；针对 KeyedStateBackend，Flink 提供了两种实现，分别是HeapKeyedStateBackend 和 RocksDBKeyedStateBackend，所以本质上 SCS 是将这两种 KeyedStateBackend 结合起来，实现了 AdaptiveKeyedStateBackend。

上图是从 HeapStateBackend 切换到 RocksDBStateBackend 的示意图；首先Flink为了支持作业并行度动态调整，内部按照 KeyGroup 粒度对状态进行划分，KeyGroup 可以通俗地理解为桶，默认有128个，可以通过 maxParallelism 参数设置，每个 Operator 自身有个并行度，代表其运行时会有多少个 SubTask，这里的例子是4，每个 SubTask 会分配到一个 KeyGroupRange，也就是一系列的桶，这里每个 SubTask 分配到32个 KeyGroup(桶)，分别负责一部分状态存储。在 SubTask 中，一个类型的状态对应一个 StateTable，StateTable 会有多个 StateTableEntry，StateTableEntry 保存真正的 key-value 状态数据；当TaskManager 接收到来自 JobManager 的 StartMigration 信号时，4个 SubTask 并行启动 StateBackend 切换，顺序遍历 StateTable 的 StateTableEntry，并写入 RocksDB 对应的 ColumnFamily。

以上是 SCS 在监控完善方面第二阶段的部分工作，相比第一阶段，基于经验规则的监控系统能够解决一些运维问题，包括阈值预警及在线作业优化，大大提高线上作业的稳定性，但经验规则有其局限性，无法解决复杂异常预测及业务负载预测等场景，针对这些新问题，SCS 进一步升级，提供了更加智能的监控系统。

智能化的监控系统

左下图是一个 TaskManager 的内存曲线图，一开始是相对有规律的波动，后来是斜线上升，简单的规则并不能识别到此类异常，所以我们引入了一些机器学习算法，主要是统计与无监督，再加上分类，通过离线模型训练及在线预测，实时根据内存 Metric 预测作业异常的概率；另外一个是资源的在线弹性伸缩，我们需要根据历史负载预测下阶段的业务负载，以便更好地评估新的资源用量，本质上是基于时间序列的回归问题。目前这部分工作还在继续，也是运维监控方面的一个重点发展方向。

展望规划

在运维监控方面，SCS 的中短期计划主要有3个；首先是完善 Metrics 指标系统，包括增加 Metric 辅助问题定位以及优化 Flink 原生 Metric 实现，同时在监控系统中增加对各种 Metrics 的深度分析，加入更多的机器学习算法预测潜在的问题，打造更加智能化的监控系统；其次是提供自动化的在线弹性伸缩能力，实时跟踪预测业务负载，自动进行在线低延时动态扩缩容；最后是完善作业日志的实时收集和分析，与现有 Metric 分析形成互补，实现全方位监控。