关于flink:Apache-Flink-在移动云实时计算的实践

摘要：本文整顿自挪动软件开发工程师谢磊在 Flink Forward Asia 2021 平台建设专场的演讲。本篇内容次要分为四个局部：

1. 实时计算平台建设
2. 中移信令业务优化
3. 稳定性实际
4. 将来方向的摸索

点击查看直播回放 & 演讲 PDF

中移（苏州）软件技术有限公司是中国移动通信有限公司的全资子公司，公司定位为中国移动云设施的构建者、云服务的提供者、云生态的绘制者。公司以挪动云为经营核心，产品和服务在电信、政务、金融、交通等畛域都有广泛应用。

一、实时计算平台介绍

实时计算引擎在挪动云的演进分为几个阶段：

• 2015 年到 16 年，咱们应用的是第一代实时计算引擎 Apache Storm；
• 17 年咱们开始调研 Apache Spark Streaming，它能够与自研框架进行整合，升高了运维压力和保护老本；
• 18 年，用户对云计算的需要越来越多，Storm 和 Spark 曾经无奈很好地满足业务。同时咱们钻研了流计算比拟闻名的几篇文章，发现 Apache Flink 曾经比拟残缺地具备了文中提到的一些语义；
• 19 年 – 20 年，咱们开始实现云服务，并把实时计算平台上线至私有云和公有云；
• 20 年 – 21 年，咱们开始调研实时数仓，并将 LakeHouse 上线挪动云。

目前 Flink 次要用于中移信令数字的解决、实时用户画像和埋点、实时数仓、实时运维监控、实时举荐以及挪动云的数据管道服务。

中移的实时计算平台性能分为三大部分。

• 第一局部是服务治理，反对了工作生命周期的托管、Flink 和 SQL 作业、Spark Streaming 作业以及引擎多版本的反对；
• 第二局部是 SQL 的反对，提供了在线 Notebook 编写、SQL 语法检测、UDF 治理和元数据管理；
• 第三局部是工作运维，反对实时工作的日志检索、实时性能指标采集以及音讯提早报警和工作反压报警等。

本文次要分享两个外围设计：引擎多版本的设计和实时工作日志检索。

在日常有工作场景中，咱们发现用户程序调试老本比拟高，用户尝试新版本引擎的周期也比拟长，此外无奈躲避用户 hack 引擎的性能以及有些工作运行失败然而没有异样信息，因而咱们引入了引擎多版本设计。

多版本提交的流程如下：用户的工作首先会提交到 rtp 服务，rtp 服务将用户程序上传到 HDFS 保留，须要提交的时候再从 HDFS 拉回来提交到 Yarn 集群。此类工作存在一个共性——作业中蕴含 Apache Flink 的外围包，这会导致很多问题。

因而，首先咱们会与业务沟通，使作业包外面不蕴含 Flink 的 core 包，然而这样的收益比拟小，所以咱们在平台侧做了一次检测，在用户在上传 jar 包的过程中被动检测用户包里是否蕴含 core 包。如果发现作业蕴含了非法外围包，则会阻止用户提交。

如此简略的操作，却为公司带来了很大的收益：

• 第一，极大升高了一些低价值 bug 的定位老本；
• 第二，作业降级和回退版本更加不便；
• 第三，进步了作业的稳定性和安全性。

在日常业务场景中，咱们须要通过日志检索来验证流程的简单逻辑。此外，原生 TM 的 UI 日志打不开，容易卡死。以及 TM UI 不反对检索，如上图所示，当业务逻辑非常复杂的时候，Flink UI 无奈提供以上性能。因而咱们设计了实时工作日志检索性能。

实时工作日志检索的设计上须要思考以下几个问题：如何采集作业程序日志，并将 TM 散布在不同的机器上？如何不侵入作业进行采集日志？如何限度作业打印大量无用日志？

• 针对第一个问题，咱们采纳的 push 模式来升高采集日志的压力；
• 针对第二个问题，参考 spring 中的 AOP 机制，咱们应用 AspectJWeaver，切入点是 log4j 的 input 或 event，之后把日志发送到 Sender；
• 针对第三个问题，咱们采纳的是 RateLimiter 来进行限流。

上图是实时工作日志检索的整体设计。咱们在原生的 TaskManager 上面加了 AOP 层，日志会先通过 TaskManager 发送 task，再发送到 AOP。整个 AOP 对用户无感知，因为采纳了切面的形式。之后再发送到 RateLimiter，再到 Sender，由 RateLimiter 进行限流的操作。接着日志持续发送到 Kafka，做检索的时候日志会被发送到 Elestic Search。

有了实时工作日志检索之后，业务程序不须要做任何改变就能够反对日志的检索。同时，开发人员能够便捷地验证业务逻辑。得益于限流措施，也不会存在日志存储瓶颈。此外，也加重了平台治理的压力。

二、中移信令业务优化

中国移动信令业务的呈现是为了解决各级政府部门有对于移动用户资源数据的需要，包含游览部门、应急部门、交通行业等，如交通布局、交通考察、游览景区等重点区域的人口流量监测、流动人口监测治理等等。

依赖于中国移动手机用户的高覆盖率，利用挪动通信网络区域服务技术以及 GIS 技术，通过对移动用户信令数据的统计，对城市人口数量、流动性等因素进行剖析预测，为城市规划、交通布局、治理、资源配置、外来人口治理、政策制订等政府治理行为提供决策数据反对。

业务日均数据大略是 10PB，20 万亿 / 天，单条数据大小 0.5KB，蕴含了 2345G 上网数据、地位信令、省份城市、网络类型、接口类型等等。数据处理也比较复杂，要做数据加密、压缩以及版本的对立等。上图是解决信令数字时的条件和业务逻辑等。

将需要化繁为简，应答到集群上，就是一个上报网关。它会将各地的信令数据进行上传，由 Flume 集群进行数据接管，再传输到 Hadoop 集群。上图能够看到，Flume 与 Hadoop 之间存在一面物理墙。

随着数据量增大，咱们也遇到了很多问题：

• 第一，Flume 集群会始终报警提醒 Flume channel full；
• 第二，防火墙超限，也会进行报警；
• 第三，Flume 在写 Kafka 的时候，Kafka 发送端会发送超时报警；
• 第四，上游解决信令数据的时候，Spark Streaming 解决是不稳固的。

上述问题总结起来能够分为两大类：

• 第一类是写入性能问题。Kafka 在写入的时候频繁超时，生产性能存在瓶颈。以及 Flume 在发送数据时无奈达到网卡的下限速度；
• 第二类是架构设计问题。架构波及的组件比拟多导致保护的老本比拟高；此外，组件职责不清晰，比方 Flume 中存在数据荡涤的逻辑；还有 Spark 逻辑和解决逻辑简单，存在多处 shuffle，解决性能不稳固。

首先要解决的是 PRO 写入 Kafka 超时的问题。为了解决这个问题，咱们进行了以下优化：

• 优化了防火墙端口；
• 优化了 Kafka 服务器的一些性能参数；
• 在 Kafka 服务器端进行了一些性能参数调优。

然而这并不能彻底解决 Flume 写入 Kafka 超时的问题，于是咱们把重点聚焦到客户端。首先是客户端的参数如何优化，尤其是 batch.size、buffer.memory 和 request.time.out 如何调优。其次是如何达到单机网络最大数网速，即单机状况下设置多少客户端并发适合。

通过实际咱们发现，当 batch.size 为 256 兆，buffer.memory 为 128 兆时，性能会达到最优，但此时并没有达到网卡的最大速度。

于是咱们进行了第二轮测试，减少了 compression.type，冀望通过压缩发送的数据来进步发送带宽，然而后果并不合乎咱们的冀望。

这是因为 Kafka 在低版本的时候存在一个问题，参数在它的验证脚本里的每个值都是一样的，所以它的压缩比会比拟大。然而理论的生产环境中每条数字都是不一样的，所以压缩比十分小。

另外一个问题是如何达到网卡的最大速度？最简略的形式是减少并行度，然而并行度并不是越大越好。通过实际发现，并发度为 4 的时候能达到网卡的最大速度，超过 4 当前均匀耗时会明显增加，也会导致 Kafka 写入超时。

第二点是 Flume channel full 的问题。

扩大服务的时候，服务的事务 API 解决是比拟底层的，须要手动进行解决。此外服务的事务处理数据的时候，须要将数据进行拷贝。如上图所示，当数据从 source 发送到 channel 的时候，会把一份数据先 copy 到内存里，从 channel 再发送到 sink 的时候，又会从 channel 再 copy 到内存。这个过程中的两次 copy 节约了资源。而 Flink 做事务的时候是借助于状态治理，因此它的解决性能是比较稳定的。另外，Flink 领有丰盛的 source 和 sink，扩展性比拟强。

因而，咱们决定应用 Flink 代替 Flume 来解决问题。替换成 Flink 当前，晋升了采集性能，解决了海量数据发送性能瓶颈，稳定性显著进步。同时，明确了组件职责，咱们将原有的服务中存在的逻辑全副转移至后端实时数据合成，让采集层专一于数据汇聚，解决层专一于数据分拣。另外，咱们对立了技术栈，端到端采纳了 Flink 框架，取得了更高的性能，也升高了开发和运维老本。

最终整体性能晋升了 1/3 且升高了保护老本。

三、稳定性实际

作业稳定性次要指服务故障以及解决计划，服务故障次要包含作业运行失败、作业生产提早、作业呈现 OOM 以及作业异样重启。对应的解决计划是能够将作业进行物理隔离，服务进行降级，增强资源监控以及对服务进行拆分。

而平台保护人员最关怀的是整体性的问题。

如果 ZooKeeper 集群中有一台服务器呈现了网络服务瞬断，它也会引起大批量的工作重启。Flink JobManager 会通过 ZooKeeper 来进行 leader 的选举和发现 CheckpointID 的计数器治理。

于是咱们剖析了 ZooKeeper 网络状态的转换。客户端在连贯 ZooKeeper 集群的时候，它的状态先是 connected 状态，网络瞬断后它会变成 Suspended 状态，Suspended 状态会转换为 lost 状态，还会持续转换为 reconnected 状态。Flink 在应用 ZooKeeper 的时候会依赖一个 curator2.0 组件，然而这个组件存在一个缺点，遇到 Suspended 状态就会间接将 leader 抛弃，这会导致大部分作业进行重启，这对于咱们的业务来说是不可承受的。

官网直到 Flink 1.14 版本才对此问题进行修复。在之前的版本下，须要从新写 LeaderLatch，同时如果应用的是 Flink 1.8 版本，还须要同时批改 ZooKeeperCheckpointIDCounter。

四、将来方向的摸索

将来，咱们次要会在这两个方向进行继续摸索：

• 第一，资源利用方向。包含 Elastic Scaling 调研和 K8s Yunikorn 资源队列调研。咱们发现 Flink 上云之后存在着资源队列的问题，所以须要将用户的资源进行分队列治理；
• 第二，数据湖方向。首先是对立流批服务网关，做实时数仓的时候可能会采纳不同的引擎，比方 Flink 和 Spark，它们属于两套不同的服务，所以须要做对立流批的服务网关。其次是数据血统、数据资产和数据品质服务化。

点击查看直播回放 & 演讲 PDF

更多 Flink 相干技术问题，可扫码退出社区钉钉交换群
第一工夫获取最新技术文章和社区动静，请关注公众号～