关于flink:Apache-Flink-在京东的实践与优化

简介： Flink 助力京东实时计算平台朝着批流一体的方向演进。
本文整顿自京东高级技术专家付海涛在 Flink Forward Asia 2020 分享的议题《Apache Flink 在京东的实际与优化》，内容包含：

业务演进和规模
容器化实际
Flink 优化改良
将来布局

一、业务演进和规模

1. 业务演进

京东在 2014 年基于 storm 打造了第一代流式解决平台，能够较好的满足业务对于数据处理实时性的要求。不过它有一些局限性，对于那些数据量特地大，然而对提早却不那么敏感的业务场景，显得有些力不从心。于是咱们在 2017 年引入了 Spark streaming，利用它的微批处理来应答这种业务场景。

随着业务的倒退和业务规模的扩充，咱们迫切需要一种兼具低提早和高吞吐能力，同时反对窗口计算、状态和恰好一次语义的计算引擎。

于是在 2018 年，咱们引入了 Flink，同时开始基于 K8s 进行实时计算容器化的降级革新；

到了 2019 年，咱们所有的实时计算工作都跑在 K8s 上了。同年咱们基于 Flink 1.8 打造了全新的 SQL 平台，不便业务开发实时计算利用；

到了 2020 年，基于 Flink 和 K8s 打造的全新实时计算平台曾经比较完善了，咱们进行了计算引擎的对立，同时反对智能诊断，来升高用户开发和运维利用的老本和难度。在过来，流解决是咱们关注的一个重点。同年，咱们也开始反对批处理，于是整个实时计算平台开始朝着批流一体的方向演进。

2. 业务场景

京东 Flink 服务于京东外部十分多的业务线，次要利用场景包含实时数仓、实时大屏、实时举荐、实时报表、实时风控和实时监控，当然还有其余一些利用场景。总之，实时计算的业务需要，个别都会用 Flink 进行开发。

3. 业务规模

目前咱们的 K8s 集群由 5000 多台机器组成，服务了京东外部 20 多个一级部门。目前在线的流计算工作数有 3000 多，流计算的解决峰值达到 5亿条每秒。

二、容器化实际

上面分享一下容器化的实际。

在 2017 年，京东外部的大多数工作还是 storm 工作，它们都是跑在物理机上的，同时还有一小部分的 Spark streaming 跑在 Yarn 上。不同的运行环境导致部署和运维的老本特地高，并且在资源利用上有肯定的节约，所以咱们迫切需要一个对立集群资源管理和调度零碎，来解决这个问题。

通过一系列的尝试、比照和优化，咱们抉择了 K8s。它不仅能够解决部署运维、资源利用的一些问题，还具备云原生弹性自愈、人造容器残缺隔离、更易扩大迁徙等长处。于是在 2018 年初，咱们开始进行容器化的降级革新。

在 2018 年的 6.18，咱们只有 20% 的工作跑在 K8s 上；到了 2019 年 2 月份，曾经实现了实时计算的所有工作都跑在 K8s 上。容器化后的实时计算平台经验了 6.18，双 11 屡次大促，扛住了洪峰压力，运行的十分稳固。

然而，咱们过来的 Flink 容器化计划是基于资源事后调配的动态形式，不能满足很多业务场景，于是咱们在 2020 年也进行了一个容器化计划的降级，前面会具体介绍。

容器化带来十分多的收益，这里次要强调三点：

第一，能够很不便的实现服务的混合部署，极大地晋升资源共享能力，节俭机器资源。

第二，人造的弹性扩大，肯定的自愈能力，并且它能够做到一个更残缺的资源隔离，更好的保障业务的稳定性。

第三，通过容器化实现了开发、测试、生产的统一环境，同时进步了部署和自动化运维的能力，使治理和运维的老本升高了一半。

咱们过来的容器化计划是基于 K8s deployment 部署的 Standalone Session 集群。它须要用户在平台创立集群时，当时预估出集群所需资源，比方须要的 jobmanager 和 taskmanager 的资源规格和个数，而后平台通过 K8s 客户端向 K8s master 发出请求，来创立 jobmanager 的 deployment 和 taskmanager 的 deployment。

其中，整个集群的高可用是基于 ZK 实现；状态存储次要是存在 HDFS，有小局部存在 OSS；监控指标 (容器指标、JVM 指标、工作指标) 上报到 Prometheus，联合 Grafana 实现指标的直观展现；日志是基于咱们京东外部的 Logbook 零碎进行采集、存储和查问。

在实践中发现，这个计划有两点有余：

第一，资源须要提前调配，无奈满足灵便多变的业务须要，无奈做到按需分配。

第二，极其场景下 Pod 不能失常拉起， 影响工作复原。

于是咱们进行了一个容器化计划的降级，实现了基于 K8s 的动静的资源分配形式。在集群创立的时候，首先咱们会依据用户指定的 job manager 的数量创立 jobmanager 的 deployment；用户在提交工作的时候，咱们会依据工作所须要的资源数，动静的向平台申请资源，创立 taskmanager。

在运行过程中，如果发现这个工作须要扩容，job manager 会和平台交互，进行动静扩容；而在发现资源节约时，会进行缩容。通过这样一个形式能够很好的解决动态预调配带来的问题，并进步了资源利用率。

此处，通过平台与 K8s 交互进行资源的创立&销毁，次要基于 4 点思考：

• 保障了计算平台对资源的监管。
• 防止了平台集群配置 & 逻辑变动对镜像的影响。
• 屏蔽了不同容器平台的差别。
• 平台原有 K8s 交互相干代码复用。

另外，为了兼容原有 Slot 调配策略 (按 slot 扩散)，在提交工作时会预估出工作所需资源并一次性申请，同时依照肯定的策略进行期待。等到有足够的资源，能满足工作运行的需要时，再进行 slot 的调配。这样很大水平上能够兼容原有的 slot 扩散调配策略。

三、Flink 优化改良

上面介绍一下 Flink 的优化改良。

1、预览拓扑

在业务应用平台的过程中，咱们发现有几个业务痛点：

第一，工作调优繁琐。在平台提交工作、运行之后如果要调整工作并行度、Slot 分组、Chaining 策略等，须要从新批改程序，或者通过命令行参数配置的形式进行调优，这是十分繁琐的。

第二，SQL 工作无奈灵便指定算子配置。

第三，工作提交到集群之后，到底须要多少资源，工作所需 Slot 数事后不分明。

第四，并行度调整后网络 buffer 有余。

为了解决这些问题，咱们开发了预览拓扑的性能：

第一，拓扑配置。用户提交工作到平台之后，咱们会把拓扑给预览进去，容许它灵便的配置这些算子的并行度。

第二，槽位分组预览。咱们会清晰的显示出工作的槽位分组状况和须要多少个槽。

第三，网络 Buffer 预估。这样能够最大限度的不便用户在平台进行业务的调整和调优。

上面简略介绍预览拓扑的工作流程。用户在平台提交 SQL 作业或 Jar 作业，这个作业提交之后，会生成一个算子的配置信息，再反馈到咱们平台。咱们平台会把整个拓扑图预览进去，而后用户就能够在线进行算子配置信息的调整。调整完之后，把调整完的配置信息从新提交到咱们平台。并且，这个过程能够是间断调整的，用户调整完感觉 ok 了就能够提交工作。提交工作之后，整个在线调整的参数就失效了。

这里工作能够屡次提交，如何保障前后两次提交生成算子稳固的对应关系呢？咱们采纳这样一个策略：如果你指定了 uidHash 或者 uid，咱们就能够拿 uidHash 和 uid 作为这样一个对应关系的 Key。如果没有，咱们会遍历整个拓扑图，依照广度优先的程序，依据算子在拓扑图中的地位生成确定的惟一的 ID。拿到惟一的 ID 之后，就能够失去一个确定的关系了。

2、背压量化

上面介绍一下咱们的第二个改良，背压量化。目前观测背压有两种形式：

第一种形式是通过 Flink UI 的背压面板，能够十分直观的查看以后的背压状况。然而它也有些问题：

• 第一，有的场景下采集不到背压。
• 第二，无奈跟踪历史背压状况。
• 第三，背压影响不直观。
• 第四，在大并行度的时候背压采集会有肯定的压力。

另外一种观测背压的形式是基于 Flink Task Metrics 指标。比如说，它会上报 inPoolUsage、outPoolUsage 这些指标，而后把它采集到 Prometheus 进行一个查问，这种形式能够解决背压历史跟踪的问题。不过它有其余一些问题：

• 第一，不同 Flink 版本的背压指标含意有肯定差别。
• 第二，剖析背压有肯定门槛，你须要对整个背压相干的指标有比拟深的意识，联结进行剖析。
• 第三，背压的影响不是那么直观，很难掂量它对业务的影响。

针对这个问题，咱们的解决方案是采集背压产生的地位、工夫和次数指标，而后上报下来。将量化的背压监控指标与运行时拓扑联合起来，就能够很直观的看到背压产生的影响 (影响工作的地位、时长和次数)。

3、文件系统反对多配置

上面介绍下文件系统反对多配置的性能。

目前在 Flink 中应用文件系统时，会应用 FileSystem.get 传入 URI，FileSystem 会将 shceme+authority 作为 key 去查找缓存的文件系统，如果不存在，依据 scheme 查找到 FileSystemFactory 调用 create 创立文件系统，返回之后就能够对文件进行操作了。不过，在平台实际过程中，常常会遇到这样的问题：

第一， 如何把 checkpoint 写入公共 HDFS，把业务数据写入另外的 HDFS？比方在平台对立治理状态，用户不关注状态的存储，只关注本人业务数据读写 HDFS 这样的场景，会有这样的需要。怎么满足这样的一个业务场景呢？

一个计划是能够把多个 HDFS 集群的配置进行交融，然而它会有个问题。就是如果多个 HDFS 集群配置有抵触的话，合并会带来肯定的问题。

另外，能够思考一些联邦的机制，比方 ViewFs，但这种机制可能又有点重。是否有其它更好的计划呢？

第二， 如何将数据从一个 OSS 存储读出、解决后写到另外一个 OSS 存储？

这两个问题都波及到如何让 Flink 的同一个文件系统反对多套配置。咱们的解决方案是通过应用不同的scheme指定和隔离不同的配置。以 HDFS 反对多配置为例，如下图所示：

第一步，在配置中设置自定义 scheme (aaHDFS) 的绑定的 scheme (HDFS) 及对应 HDFS 配置门路。

第二步，在调用 FileSystem.get 时，从 aaHDFS 对应的门路加载 Hadoop 配置。

第三步，在读写 HDFS 时，应用 HadoopFileSystemWrapper 将用户自定义 scheme 的门路 (aaHDFS://) 转换为实在的 hadoop 门路 (HDFS://)。

咱们也做了许多其它的优化和扩大，次要分为三大块。

• 第一块是性能的优化，包含 HDFS 优化 (合并小文件、升高 RPC 调用)、基于负载的动静 rebalance、Slot 调配策略扩大 (程序、随机、按槽扩散) 等等。
• 第二块是稳定性的优化，包含 ZK 防抖、JM Failover 优化、最初一次 checkpoint 作为 savepoint 等等。
• 第三块是易用性的优化，包含日志加强 (日志拆散、日志级别动静配置)、SQL 扩大 (窗口反对增量计算，反对offset)、智能诊断等等。

四、将来布局

最初是将来布局。演绎为 4 点：

• 第一，继续欠缺 SQL 平台。继续加强欠缺 SQL 平台，推动用户更多地应用 SQL 开发作业。
• 第二，智能诊断和主动调整。全自动智能诊断，自适应调整运行参数，作业自治。
• 第三，批流一体。SQL 层面批流一体，兼具低提早的流解决和高稳固的批处理能力。
• 第四，AI 摸索实际。批流对立和 AI 实时化，人工智能场景摸索与实际。

原文链接
本文为阿里云原创内容，未经容许不得转载。