分享嘉宾&＃xff1a;付海涛 京东 高级技术专家

编辑整理&＃xff1a;刘明明

来源:微信公众号:DataFunTalk

出处:https://mp.weixin.qq.com/s?__biz&＃61;MzU1NTMyOTI4Mw&＃61;&＃61;&mid&＃61;2247515115&idx&＃61;1&sn&＃61;4a66938e452ba489cab72171d3ec902a

导读&＃xff1a; Flink是目前流式处理领域的热门引擎&＃xff0c;具备高吞吐、低延迟的特点&＃xff0c;在实时数仓、实时风控、实时推荐等多个场景有着广泛的应用。京东于2018年开始基于Flink&＃43;K8s深入打造高性能、稳定、可靠、易用的实时计算平台&＃xff0c;支撑了京东内部多条业务线平稳度过618、双11多次大促。本次讲演将分享京东Flink计算平台在容器化实践过程中遇到的问题和方案&＃xff0c;在性能、稳定性、易用性等方面对社区版Flink所做的深入的定制和优化&＃xff0c;以及未来的展望和规划。主要内容包括&＃xff1a;

• 实时计算演进
• Flink容器化实践
• Flink优化改进
• 未来规划

01

实时计算引进

1.发展历程

最初大数据的模式基本都是T&＃43;1&＃xff0c;但是随着业务发展&＃xff0c;对数据实时性的要求越来越高&＃xff0c;比如对于一个数据&＃xff0c;希望能够在分钟级甚至秒级得到计算结果。京东是在2014年开始基于Storm打造第一代流式计算平台&＃xff0c;并在Storm的基础上&＃xff0c;做了很多优化改进&＃xff0c;比如基于cgroup实现对worker使用资源的隔离、网络传输压缩优化、引入任务粒度toplogy master分担zk压力等。到2016年&＃xff0c;Storm已经成为京东内部流式处理的最主要的计算引擎&＃xff0c;服务于各个业务线&＃xff0c;可以达到比较高的实时性。

随着业务规模的不断扩大&＃xff0c;Storm也暴露出许多问题&＃xff0c;特别是对于吞吐量巨大、但是对于延迟不是那么敏感的业务场景显得力不从心。于是&＃xff0c;京东在2017年引入了Spark Streaming流式计算引擎&＃xff0c;用于满足此类场景业务需要。

随着业务的发展&＃xff0c;不光是对于数据的延迟有很高要求&＃xff0c;同时对于数据的吞吐处理能力也有很高的要求&＃xff0c;所以迫切需要一个兼具低延迟和高吞吐能力的计算框架&＃xff0c;于是在2018年我们引入了Flink。在Flink社区版的基础上&＃xff0c;我们从性能、稳定性、易用性还有功能等方面&＃xff0c;都做了一些深入的定制和优化。同时我们基于k8s实现了实时计算全面的容器化&＃xff0c;因为容器化有很多的优点&＃xff0c;它可以做到很好的资源隔离&＃xff0c;同时它有一个很强的自愈能力&＃xff0c;另外它很容易实现资源的弹性调度。同时我们基于Flink打造了全新的SQL平台&＃xff0c;降低用户开发实时计算应用的门槛。

到2020年&＃xff0c;基于Flink和k8s实时计算平台已经做的比较完善了。过去流式处理是我们关注的重点&＃xff0c;今年我们也开始逐渐的支持批处理&＃xff0c;朝着批流一体的方向演进。另外AI是目前比较火的一个方向&＃xff0c;对于AI来说&＃xff0c;它的实时化也是一个重要的研究方向。所以我们的实时计算平台将会朝着批流一体和AI的方向进行发展。

2.平台架构

上面是京东实时计算平台JRC的整体架构&＃xff0c;整个架构以定制化改造后的Flink为核心&＃xff0c;Flink运行在K8S上&＃xff0c;状态存储在HDFS集群上&＃xff0c;通过Zookeeper保证集群的高可用。 支持流式源JDQ(京东基于Kafka深入定制实现的实时数据总线)和Hive&＃xff0c;数据主要写入JimDB(京东内存数据库)、ES、Hbase和京东OLAP。计算平台支持SQL和普通JAR包两种方式的作业&＃xff0c;具有配置、部署、调试、监控、和日志处理等功能。

3. 业务场景

京东Flink服务于京东内部非常多的业务线&＃xff0c;有70多个一级部门在使用&＃xff0c;主要应用场景包括实时数仓&＃xff0c;实时大屏&＃xff0c;实时推荐&＃xff0c;实时报表&＃xff0c;实时风控和实时监控&＃xff0c;当然还有其他一些应用场景。对数据计算实时性有一定要求的场景&＃xff0c;一般都会使用Flink进行开发。

4. 业务规模

京东Flink集群目前由5000多台物理机组成&＃xff0c;它服务了京东内部70多个一级业务部门&＃xff0c;目前线上的流计算任务大概有3000多个&＃xff0c;数据的处理能力可以达到每分钟数十亿甚至更高。

02

Flink容器化实践

1.容器化历程

京东从2018年开始进行计算引擎的容器化改造&＃xff0c;2019年初已经实现计算单元全部容器化&＃xff0c;2020年进行了容器化方案升级&＃xff0c;使用native k8s实现计算资源的弹性扩容。容器化改造的好处是提升了资源使用率&＃xff0c;提高了研发效率&＃xff0c;增强了业务稳定性&＃xff0c;减少了运维部署成本。

2.容器化方案

旧的容器化方案是基于k8s Deployment部署的Standalone Session集群&＃xff0c;它需要事先预估出集群所需资源&＃xff0c;比如需要的JobManager和TaskManager的资源规格和个数。然后JRC平台通过K8S客户端向K8S Master提出请求&＃xff0c;创建JobManager的Deployment和TaskManager的Deployment。其中使用ZK保证高可用&＃xff0c;使用Hdfs实现状态存储&＃xff0c;使用Prometheus实现监控指标的上传&＃xff0c;结合Grafana实现指标的直观展示。集群使用ES存储日志&＃xff0c;方便日志的查询。

3.容器化遇到的问题&对策

容器化过程中可能遇到很多问题&＃xff1a;

① JM/TM故障自动恢复

应用部署在容器中&＃xff0c;当应用出现异常时&＃xff0c;如何发现应用或者异常的情况呢&＃xff1f;比如可以使用存活探针&＃xff0c;编写检测脚本定期读取应用的心跳信息。当检测到Pod处于不健康状态时&＃xff0c;可以采用k8s的重启机制来重启不健康的容器。

② 减少Pod异常对业务影响

在k8s中由于硬件异常、资源过载、Pod不健康等问题会导致Pod被驱逐或自动重启&＃xff0c;Pod重启时势必会影响到该Pod上分布计算任务的正常运行。这个时候可以考虑采用适当的重启策略、改造内核等方案来减少对任务影响。比如京东实现了JM Failover优化&＃xff0c;当Pod异常引起JM Failover时采用的是任务不恢复、重建任务状态恢复的方式&＃xff0c;可以一定程度上减少Pod重启对业务带来的影响。

③ 性能问题

在容器环境下&＃xff0c;JVM对cpu和内存的感知会有一定的问题&＃xff0c;在Java8版本中&＃xff0c;一些参数就要进行显式的设置。对于机器性能差异或热点等问题导致部分Pod计算慢的问题&＃xff0c;可以考虑进行针对性优化(比如实现基于负载的数据分发)或处理(比如检测到计算慢的Pod将其驱逐到负载较低的机器)。此外&＃xff0c;对于使用容器网络的情况下&＃xff0c;可能会带来一定的网络性能损耗&＃xff0c;此时可以根据情况选择使用主机网络避免网络虚拟化带来的开销&＃xff0c;或者选择更高性能的网络插件。

④ 重要业务稳定性

如何保证业务的稳定性是一个需要重点考虑的问题。除了保证系统各个环节的高可用外&＃xff0c;还可以根据业务情况考虑使用其它合理的方案&＃xff0c;例如业务分级管理&＃xff0c;独立资源池&＃xff0c;多机房互备等。

4.容器化方案升级(Native k8s)

原有容器化方案存在一定的问题&＃xff1a;

• 资源需要提前分配
• 无法实现资源弹性伸缩
• 极端场景下Pod不能正常拉起&＃xff0c;影响任务恢复
• 重要业务稳定性

容器化升级的解决方案是采用Native K8s的方式。由JRC平台先向K8S Master发出请求&＃xff0c;创建JobManager的Deployment&＃xff1b;然后在用户通过Rest服务提交任务后&＃xff0c;由JobMaster通过JDResourceManager 向JRC平台发出请求&＃xff0c;然后JRC平台向 K8s Master 动态申请资源去创建运行TaskManager 的Pod。

此处&＃xff0c;通过引入JRC平台与K8s交互&＃xff0c;屏蔽了不同容器平台的差异&＃xff0c;解耦了镜像与平台集群配置&逻辑变化。另外&＃xff0c;为了兼容原有Slot分配策略&＃xff0c;在提交任务时会预估出任务所需资源并一次性申请&＃xff0c;之后采用等待一定时间后进行slot分配的方式达到兼容目的。

03

Flink优化改进

主要做了以下四个方面的优化&＃xff1a;

• 性能
• 稳定性
• 易用性
• 功能扩展

下边分几个重要的点进行讲解&＃xff1a;

1.预览拓扑

预览拓扑主要是为了解决业务的一些痛点&＃xff1a;比如任务调优繁琐、SQL任务无法指定并行度、任务需要的额Slot数不清楚、并行度调整后网络buffer不足等。在Flink任务调试阶段&＃xff0c;对任务并行度、Slot分组、Chaining策略的调整是个反复的过程&＃xff0c;如果把参数写到命令行就太繁琐了。而基于预览拓扑就可以很方便地对这些参数进行配置。

预览拓扑基本的实现方案如上图&＃xff1a;用户提交JAR包后可根据JAR包生成对应的拓扑图&＃xff0c;之后用户根据拓扑图可以进行在线调整&＃xff0c;最后自动将修改后的配置和原来的JAR包一起进行任务提交。

预览拓扑机制使得不修改程序多次提交任务调优成为可能&＃xff0c;但是如何保证前后两次提交生成算子稳定的对应关系呢&＃xff1f;解决方案的关键是保证算子有稳定的唯一身份标识&＃xff0c;具体算法是&＃xff1a;如果算子指定了uidHash就用uidHash&＃xff0c;如果算子指定了uid就使用uid&＃xff0c;否则就从source开始广度优先遍历&＃xff0c;利用算子在graph中的位置生成一个稳定hash值。

2.背压量化

第二个重要的优化是背压量化。

在Flink开发的时候&＃xff0c;主要有两种方式&＃xff1a;

① 通过Flink UI背压面板观察是否背压。使用这种方式在某些场景比较方便&＃xff0c;但是它存在几个问题&＃xff1a;

• 在有些场景下采集不到背压
• 对于历史背压情况无法跟踪
• 背压影响不直观
• 大并行度时背压采集压力

② 通过任务背压相关指标进行观察和分析&＃xff0c;通过将指标定期采集并存储起来&＃xff0c;可以进行实时或历史的背压分析。但是它也有一些不足的地方&＃xff1a;

• 不同Flink版本中指标含义有一定差异
• 分析背压有一定门槛&＃xff0c;需要对于指标含义有深入理解&＃xff0c;联合进行分析
• 背压未量化&＃xff0c;对业务影响程度不够直观

京东的解决方案是采集背压发生的位置、时间和次数指标&＃xff0c;并对这些指标进行上报存储。同时对量化的背压指标结合运行时拓扑&＃xff0c;可以精确反映发生背压现场的情况。

3.HDFS优化

随着业务数量的增多&＃xff0c;HDFS集群的压力就会变得很大。这会直接导致RPC响应时间变慢&＃xff0c;造成请求堆积&＃xff0c;同时大量小文件也会对NN内存造成很大压力。对此京东尝试的解决方案有4方面&＃xff1a;限制checkpoint最小间隔&＃xff0c;时间最小设置在1min左右可以满足大部分业务需求&＃xff1b;进行小文件合并&＃xff1b;降低cp创建和删除时的hdfs rpc请求&＃xff1b;HDFS集群多ns分散均衡压力。

4.网络分发优化

在实践过程中我们发现&＃xff0c;即使业务使用了rebalance并且对任务进行了打散分布&＃xff0c;但是由于机器处理能力和负载的差异&＃xff0c;会导致任务各个并行度不同程序的背压表现&＃xff0c;严重影响了任务的性能。为此&＃xff0c;我们开发了基于负载的动态rebalance&＃xff0c;在数据进行分发时优先选择下游负载最小的channel进行分发。

经测试&＃xff0c;在特定场景下性能能够提升近一倍。

5.ZK防抖

目前一般都是使用ZK集群实现Flink集群的高可用&＃xff0c;但是当网络抖动、机器繁忙、ZK集群暂时无响应或运维机器的时候&＃xff0c;都可能会导致任务重启。

任务重启的原因是由于在这些场景发生时&＃xff0c;Curator会将状态设置为suspended&＃xff0c;并且Curator认为suspended为Error状态&＃xff0c;从而会释放leader&＃xff0c;Flink发现notleader后会revokeLeadership&＃xff0c;从而造成任务重启。

一个可行的解决办法是升级Curator的版本&＃xff0c;同时将connectionStateErrorPolicy设置为SessionConnetionStateErrorPolicy。

6.日志分离

目前我们一个集群是支持跑多个任务的&＃xff0c;这时日志会出现的问题是&＃xff1a;任务的日志和集群Framework日志混在一起&＃xff0c;同时集群的多个任务日志也是混在一起的&＃xff0c;不太方便用户查看日志&＃xff0c;快速定位问题。

为了解决这个问题&＃xff0c;首先要弄清楚目前Flink加载日志框架的基本机制&＃xff1a; 为了避免跟业务Job中可能包含的日志框架的依赖、配置文件产生冲突&＃xff0c;Flink日志相关类的加载都代理给TaskManager框架的类加载器&＃xff0c;也就是Parent Classloader&＃xff0c;而框架加载的这些类都是从Flink安装包的lib目录下加载的。对于日志配置文件&＃xff0c;Flink通过 JVM 启动参数来指定配置日志配置文件路径。

日志分离的解决方案是&＃xff1a;将日志相关jar包加入到各个task自己classloader(user classloader)的类路径中&＃xff1b;同时确保使用user classloader加载日志类和加载自己的日志配置&＃xff1b;

另外对于使用了Flink框架的类(比如PrintSinkFunction)&＃xff0c;日志不能做到很好的分离&＃xff0c;可以考虑使用logback MDC机制。

04

未来规划

未来规划主要包括四个方面:

① 统一计算引擎

引擎Storm全部升级为Flink&＃xff0c;这样可以减少平台的运维成本&＃xff0c;同时可以提高作业性能(目前已经接近完成)。

② 更多SQL作业

持续完善SQL平台&＃xff0c;降低用户的使用门槛&＃xff0c;推动用户更多使用SQL开发作业。

③ 智能运维

使用智能诊断&＃xff0c;自适应调整运行参数&＃xff0c;提升任务的鲁棒性

④ 批流一体

深度打造批流一体实时计算平台&＃xff0c;兼具低延迟的流处理和高性能的批处理能力。另外统一架构&＃xff0c;实现代码复用&＃xff0c;降低用户的使用成本。

今天的分享就到这里&＃xff0c;谢谢大家。

在文末分享、点赞、在看&＃xff0c;给个3连击呗~

嘉宾介绍&＃xff1a;

付海涛

京东 | 高级技术专家

拥有多年中间件、互联网云平台和大数据开发经验&＃xff0c;对分布式计算、容器、微服务有较深入的理解。2018年加入京东&＃xff0c;主要负责实时计算引擎storm、flink的相关优化和开发工作。

分享嘉宾&＃xff1a;付海涛 京东 高级技术专家

编辑整理&＃xff1a;刘明明

来源:微信公众号:DataFunTalk

出处:https://mp.weixin.qq.com/s?__biz&＃61;MzU1NTMyOTI4Mw&＃61;&＃61;&mid&＃61;2247515115&idx&＃61;1&sn&＃61;4a66938e452ba489cab72171d3ec902a