Flink在美团的实践与应用

美团实时计算平台现状和背景

实时平台架构

上图呈现的是当前美团实时计算平台的简要架构。最底层是数据缓存层，能够看到美团测的全部日志类的数据，都是经过统一的日志收集系统收集到Kafka。Kafka做为最大的数据中转层，支撑了美团线上的大量业务，包括离线拉取，以及部分实时处理业务等。在数据缓存层之上，是一个引擎层，这一层的左侧是咱们目前提供的实时计算引擎，包括Storm和Flink。Storm在此以前是 standalone 模式的部署方式，Flink因为其如今运行的环境，美团选择的是On YARN模式，除了计算引擎以外，咱们还提供一些实时存储功能，用于存储计算的中间状态、计算的结果、以及维度数据等，目前这一类存储包含Hbase、Redis以及ES。在计算引擎之上，是趋于五花八门的一层，这一层主要面向数据开发的同窗。实时数据开发面临诸多问题，例如在程序的调试调优方面就要比普通的程序开发困难不少。在数据平台这一层，美团面向用户提供的实时计算平台，不只能够托管做业，还能够实现调优诊断以及监控报警，此外还有实时数据的检索以及权限管理等功能。除了提供面向数据开发同窗的实时计算平台，美团如今正在作的事情还包括构建元数据中心。这也是将来咱们想作SQL的一个前提，元数据中心是承载实时流系统的一个重要环节，咱们能够把它理解为实时系统中的大脑，它能够存储数据的Schema，Meta。架构的最顶层就是咱们如今实时计算平台支撑的业务，不只包含线上业务日志的实时查询和检索，还涵盖当下十分热门的实时机器学习。机器学习常常会涉及到搜索和推荐场景，这两个场景最显著特色：1、会产生海量实时数据；2、流量的QPS至关高。此时就须要实时计算平台承载部分实时特征的提取工做，实现应用的搜索推荐服务。还有一类是比较常见的场景，包括实时的特征聚合，斑马Watcher（能够认为是一个监控类的服务），实时数仓等。

以上就是美团目前实时计算平台的简要架构。

实时平台现状

美团实时计算平台的现状是做业量如今已经达到了近万，集群的节点的规模是千级别的，天级消息量已经达到了万亿级，高峰期的消息量可以达到千万条每秒。

痛点和问题

美团在调研使用Flink以前遇到了一些痛点和问题:

• 实时计算精确性问题：在调研使用Flink以前美团很大规模的做业是基于Storm去开发的，Storm主要的计算语义是At-Least-Once，这种语义在保证正确性上其实是有一些问题的，在Trident以前Storm是无状态的处理。虽然Storm
  Trident提供了一个维护状态的精确的开发，可是它是基于串行的Batch提交的，那么遇到问题在处理性能上可能会有一点瓶颈。而且Trident是基于微批的处理，在延迟上没有达到比较高的要求，因此不能知足一些对延迟比较高需求的业务。

• 流处理中的状态管理问题：基于以前的流处理过程当中状态管理的问题是很是大的一类问题。状态管理除了会影响到好比说计算状态的一致性，还会影响到实时计算处理的性能以及故障恢复时候的能力。而Flink最突出的一个优点就是状态管理。

• 实时计算表义能力的局限性：在实时计算以前不少公司大部分的数据开发仍是面向离线的场景，近几年实时的场景也慢慢火热起来了。那与离线的处理不一样的是，实时的场景下，数据处理的表意能力可能有必定的限制，好比说他要进行精确计算以及时间窗口都是须要在此之上去开发不少功能性的东西。

• 开发调试成本高：近千结点的集群上已经跑了近万的做业，分布式的处理的引擎，手工写代码的方式，给数据开发的同窗也带来了很高开发和调试的成本，再去维护的时候，运维成本也比较高。

Flink探索关注点

在上面这些痛点和问题的背景下，美团从去年开始进行Flink的探索，关注点主要有如下4个方面：

• ExactlyOnce计算能力

• 状态管理能力

• 窗口/Join/时间处理等等

• SQL/TableAPI

Flink在美团的实践

下面带你们来看一下，美团从去年投入生产过程当中都遇到了哪些问题，以及一些解决方案，分为下面三个部分：

稳定性实践

稳定性实践-资源隔离

• 资源隔离的考虑：分场景、按业务

  1. 高峰期不一样，运维时间不一样；
  2. 可靠性、延迟需求不一样；
  3. 应用场景，重要性不一样

• 资源隔离的策略：

  1. YARN打标签，节点物理隔离；
  2. 离线DataNode与实时计算节点的隔离

稳定性实践-智能调度

智能调度目的也是为了解决资源不均的问题，如今普通的调度策略就是基于CPU，基于内存去调度的。除此以外，在生产过程当中也发现了一些其余的问题，好比说Flink是会依赖本地磁盘，进行依赖本地磁盘作本地的状态的存储，因此磁盘IO，还有磁盘的容量，也是一类考虑的问题点，除此以外还包括网卡流量，由于每一个业务的流量的状态是不同的，分配进来会致使流量的高峰，把某一个网卡打满，从而影响其余业务，因此指望的话是说作一些智能调度化的事情。目前暂时能作到的是从cpu和内存两方面，将来会从其余方面作一些更优的调度策略。

稳定性实践-故障容错

• 节点/网络故障

  • JobManagerHA

  • 自动拉起

与Storm不一样的是，知道Storm在遇到异常的时候是很是简单粗暴的，好比说有发生了异常，可能用户没有在代码中进行比较规范的异常处理，可是没有关系，由于worker会重启做业还会继续执行，而且他保证的是At-Least-Once这样的语义，好比说一个网络超时的异常对他而言影响可能并无那么大，可是Flink不一样的是他对异常的容忍度是很是的苛刻的，那时候就考虑的是好比说会发生节点或者是网络的故障，那JobManager单点问题可能就是一个瓶颈，JobManager那个若是挂掉的话，那么可能对整个做业的影响就是不可回复的，因此考虑了作HA，另一个就是会去考虑一些因为运维的因素而致使的那做业，还有除此以外，可能有一些用户做业是没有开启CheckPoint，但若是是由于节点或者是网络故障致使挂掉，但愿会在平台内层作一些自动拉起的策略，去保证做业运行的稳定性。

• 上下游容错

  • FlinkKafka08异常重试

咱们的数据源主要是Kafka，读写Kafka是一类很是常见的实时流处理避不开的一个内容，而Kafka自己的集群规模是很是比较大的，所以节点的故障出现是一个常态问题，在此基础上咱们对节点故障进行了一些容错，好比说节点挂掉或者是数据均衡的时候，Leader会切换，那自己Flink的读写对Leader的切换容忍度没有那么高，在此基础上咱们对一些特定场景的，以及一些特有的异常作的一些优化，进行了一些重试。

• 容灾

  • 多机房

  • 流热备

容灾可能你们对考虑的并很少，好比说有没有可能一个机房的全部的节点都挂掉了，或者是没法访问了，虽然它是一个小几率的事件，但它也是会发生的。因此如今也会考虑作多机房的一些部署，包括还有Kafka的一些热备。

Flink平台化

Flink平台化-做业管理

在实践过程当中，为了解决做业管理的一些问题，减小用户开发的一些成本，咱们作了一些平台化的工做，下图是一个做业提交的界面展现，包括做业的配置，做业生命周期的管理，报警的一些配置，延迟的展现，都是集成在实时计算平台的。

Flink平台化-监控报警

在监控上咱们也作了一些事情，对于实时做业来说，对监控的要求会更高，好比说在做业延迟的时候对业务的影响也比较大，因此作了一些延迟的报警，包括做业状态的报警，好比说做业存活的状态，以及做业运行的状态，还有将来会作一些自定义Metrics的报警。自定义Metrics是将来会考虑基于做业处理自己的内容性，作一些可配置化的一些报警。

Flink平台化-调优诊断

• 实时计算引擎提供统一日志和Metrics方案

• 为业务提供按条件过滤的日志检索

• 为业务提供自定义时间跨度的指标查询

• 基于日志和指标，为业务提供可配置的报警

另外就是刚刚提到说在开发实时做业的时候，调优和诊断是一个比较难的痛点，就是用户不是很难去查看分布式的日志，因此也提供了一套统一的解决方案。这套解决方案主要是针对日志和Metrics，会在针对引擎那一层作一些日志和Metrics的上报，那么它会经过统一的日志收集系统，将这些原始的日志，还有Metrics聚集到Kafka那一层。从此Kafka这一层你们能够发现它有两个下游，一方面是作日志到ES的数据同步，目的的话是说可以进入日志中心去作一些日志的检索，另一方面是经过一些聚合处理流转到写入到OpenTSDB把数据作依赖，这份聚合后的数据会作一些查询，一方面是Metrics的查询展现，另一方面就是包括实作的一些相关的报警。

下图是当前某一个做业的一个可支持跨天维度的Metrics的一个查询的页面。能够看到说若是是可以经过纵向的对比，能够发现除了做业在某一个时间点是由于什么状况致使的？好比说延迟啊这样容易帮用户判断一些他的作做业的一些问题。除了做业的运行状态以外，也会先就是采集一些节点的基本信息做为横向的对比

下图是当前的日志的一些查询，它记录了，由于做业在挂掉以后，每个ApplicationID可能会变化，那么基于做业惟一的惟一的主键做业名去搜集了全部的做业，从建立之初到当前运行的日志，那么能够容许用户的跨Application的日志查询。

生态建设

为了适配这两类MQ作了不一样的事情，对于线上的MQ，指望去作一次同步屡次消费，目的是避免对线上的业务形成影响，对于的生产类的Kafka就是线下的Kafka，作了一些地址的地址的屏蔽，还有基础基础的一些配置，包括一些权限的管理，还有指标的采集。

Flink在美团的应用

下面会给你们讲两个Flink在美团的真实使用的案例。第一个是Petra，Petra实际上是一个实时指标的一个聚合的系统，它实际上是面向公司的一个统一化的解决方案。它主要面向的业务场景就是基于业务的时间去统计，还有计算一些实时的指标，要求的话是低时延，他还有一个就是说，由于它是面向的是通用的业务，因为业务多是各自会有各自不一样的维度，每个业务可能包含了包括应用通道机房，还有其余的各自应用各个业务特有的一些维度，并且这些维度可能涉及到比较多，另一个就是说它多是就是业务须要去作一些复合的指标的计算，好比说最多见的交易成功率，他可能须要去计算支付的成功数，还有和下单数的比例。另一个就是说统一化的指标聚合可能面向的仍是一个系统，好比说是一些B端或者是R段的一些监控类的系统，那么系统对于指标系统的诉求，就是说我但愿指标聚合可以最真最实时最精确的可以产生一些结果，数据保证说它的下游系统可以真实的监控到当前的信息。右边图是我当一个Metrics展现的一个事例。能够看到其余其实跟刚刚讲也是比较相似的，就是说包含了业务的不一样维度的一些指标汇聚的结果。

Petra实时指标聚合

• 业务场景：

  • 基于业务时间（事件时间）

  • 多业务维度：如应用、通道、机房等

  • 复合指标计算：如交易成功率=支付成功数/下单数

  • 低延迟：秒级结果输出

• Exactlyonce的精确性保障

  • Flinkcheckpoint机制

• 维度计算中数据倾斜

  • 热点key散列

• 对晚到数据的容忍能力

  • 窗口的设置与资源的权衡

在用Flink去作实时指标复核的系统的时候，着重从这几方面去考虑了。第一个方面是说精确的计算，包括使用了FLink和CheckPoint的机制去保证说我能作到不丢不重的计算，第一个首先是由统一化的Metrics流入到一个预聚合的模块，预聚合的模块主要去作一些初始化的一些聚合，其中的为何会分预聚合和全量聚合主要的解决一类问题，包括就刚刚那位同窗问的一个问题，就是数据倾斜的问题，好比说在热点K发生的时候，当前的解决方案也是经过预聚合的方式去作一些缓冲，让尽可能把K去打散，再聚合全量聚合模块去作汇聚。那其实也是只能解决一部分问题，因此后面也考虑说在性能的优化上包括去探索状态存储的性能。下面的话仍是包含晚到数据的容忍能力，由于指标汇聚可能刚刚也提到说要包含一些复合的指标，那么符合的指标所依赖的数据可能来自于不一样的流，即使来自于同一个流，可能每个数据上报的时候，可能也会有晚到的状况发生，那时候须要去对数据关联作晚到的容忍，容忍的一方面是说能够设置晚到的Lateness的延迟，另外一方面是能够设置窗口的长度，可是其实在现实的应用场景上，其实还有一方面考虑就是说除了去尽可能的去拉长时间，还要考虑真正的计算成本，因此在这方面也作了一些权衡，那么指标基本就是通过全量聚合以后，聚合结果会回写Kafka，通过数据同步的模块写到OpenTSDB去作，最后去grafana那作指标的展现，另外一方面可能去应用到经过Facebook包同步的模块去同步到报警的系统里面去作一些指标，基于指标的报警。

下图是如今提供的产品化的Petra的一个展现的机示意图，能够看到目前的话就是定义了某一些经常使用的算子，以及维度的配置，容许用户进行配置话的处理，直接去可以获取到他指望要的指标的一个展现和汇聚的结果。目前还在探索说为Petra基于Sql作一些事情，由于不少用户也比较就是在就是习惯上也能够倾向于说我要去写Sql去完成这样的统计，因此也会基于此说依赖Flink的自己的对SQl还有TableAPI的支持，也会在Sql的场景上进行一些探索。

MLX机器学习平台

---

第二类应用就是机器学习的一个场景，机器学习的场景可能会依赖离线的特征数据以及实时的特征数据。一个是基于现有的离线场景下的特征提取，通过了批处理，流转到了离线的集群。另一个就是近线模式，近线模式出的数据就是现有的从日志收集系统流转过来的统一的日志，通过Flink的处理，就是包括流的关联以及特征的提取，再作模型的训练，流转到最终的训练的集群，训练的集群会产出P的特征，还有都是Delta的特征，最终将这些特征影响到线上的线上的特征的一个训练的一个服务上。这是一个比较常见的，好比说比较就是通用的也是比较通用的一个场景，目前的话主要应用的方可能包含了搜索还有推荐，以及一些其余的业务。

将来展望

将来的话可能也是经过也是指望在这三方面进行作一些更多的事情，刚刚也提到了包括状态的管理，第一个是状态的统一的，好比说Sql化的统一的管理，但愿有统一的配置，帮用户去选择一些指望的回滚点。另一个就是大状态的性能优化，由于好比说像作一些流量数据的双流的关联的时候，如今也遇到了一些性能瓶颈的问题，对于说啊基于内存型的状态，基于内存型的数据的处理，以及基于RocksDB的状态的处理，作过性能的比较，发现其实性能的差别仍是有一些大的，因此但愿说在基于RocksDBBackend的上面可以去尽可能去更多的作一些优化，从而提高做业处理的性能。第二方面就是Sql，Sql的话应该是每个位就是当前可能各个公司都在作的一个方向，由于以前也有对Sql作一些探索，包括提供了基于Storm的一些Sql的表示，可是可能对于以前的话对于与语义的表达可能会有一些欠缺，因此但愿说在基于Flink可去解决这些方面的事情，以及包括Sql的并发度的一些配置的优化，包括Sql的查询的一些优化，都但愿说在Flink将来可以去优化更多的东西，去真正能使Sql应用到生产的环境。

另一方面的话就是会进行新的场景的也在作新的场景的一些探索，指望是好比说包括刚刚也提到说除了流式的处理，也指望说把离线的场景下的数据进行一些合并，经过统一的Sql的API去提供给业务作更多的服务，包括流处理，还有批处理的结合。