本文整理自京东高级技术专家付海涛在 Flink Forward Asia 2020 分享的议题《Apache Flink 在京东的实践与优化》,内容包括:
- 业务演进和规模
- 容器化实践
- Flink 优化改进
- 未来规划
GitHub 地址
https://github.com/apache/flink
欢迎大家给 Flink 点赞送 star~
一、业务演进和规模
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 实时化,人工智能场景探索与实践。
第三届 Apache Flink 极客挑战赛报名开始!
30 万奖金等你来!
伴随着海量数据的冲击,数据处理分析能力在业务中的价值与日俱增,各行各业对于数据处理时效性的探索也在不断深入,作为主打实时计算的计算引擎 - Apache Flink 应运而生。
为给行业带来更多实时计算赋能实践的思路,鼓励广大热爱技术的开发者加深对 Flink 的掌握,Apache Flink 社区联手阿里云、英特尔、阿里巴巴人工智能治理与可持续发展实验室 (AAIG)、Occlum 联合举办 "第三届 Apache Flink 极客挑战赛暨 AAIG CUP" 活动,即日起正式启动。
👉 点击了解更多赛事信息 👈
更多 Flink 相关技术问题,可扫码加入社区钉钉交流群
第一时间获取最新技术文章和社区动态,请关注公众号~
活动推荐
阿里云基于 Apache Flink 构建的企业级产品-实时计算Flink版现开启活动:
99 元试用 实时计算Flink版(包年包月、10CU)即有机会获得 Flink 独家定制T恤;另包 3 个月及以上还有 85 折优惠!
了解活动详情:https://www.aliyun.com/product/bigdata/sc