摩登3内部554258_京东Flink优化与技术实践


享嘉宾:
付海涛

东 高级技术专家


编辑整理:赵明明

出品平台:DataFunTalk

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

  • 实时计算演进

  • Flink容器化实践

  • Flink优化改进

  • 未来规划

01
实时计算引进
1.发展历程

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

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

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

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

2.平台架构


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

3. 业务场景


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


4. 业务规模


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

02
Flink容器化实践

1.容器化历程


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

2.容器化方案


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

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

容器化过程中可能遇到很多问题:

① JM/TM故障自动恢复

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

减少Pod异常对业务影响

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

性能问题

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

重要业务稳定性

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

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


原有容器化方案存在一定的问题:

  • 资源需要提前分配

  • 无法实现资源弹性伸缩

  • 极端场景下Pod不能正常拉起,影响任务恢复

  • 重要业务稳定性

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

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

03
Flink优化改进

主要做了以下四个方面的优化:

  • 性能

  • 稳定性

  • 易用性

  • 功能扩展

下边分几个重要的点进行讲解:

1.预览拓扑

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

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

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

2.背压量化


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

在Flink开发的时候,主要有两种方式:

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

  • 在有些场景下采集不到背压

  • 对于历史背压情况无法跟踪

  • 背压影响不直观

  • 大并行度时背压采集压力

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

  • 不同Flink版本中指标含义有一定差异

  • 分析背压有一定门槛,需要对于指标含义有深入理解,联合进行分析

  • 背压未量化,对业务影响程度不够直观

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

3.HDFS优化


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

4.网络分发优化


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

经测试,在特定场景下性能能够提升近一倍。

5.ZK防抖


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

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

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

6.日志分离


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

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

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

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

04
未来规划

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

统一计算引擎

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

更多SQL作业

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

智能运维

长按订阅更多精彩▼

免责声明:本文内容由21ic获得授权后发布,版权归原作者所有,本平台仅提供信息存储服务。文章仅代表作者个人观点,不代表本平台立场,如有问题,请联系我们,谢谢!