关于flink:快手基于-Apache-Flink-的优化实践

本次由快手刘建刚老师分享，内容次要分为三局部。首先介绍流式计算的基本概念，而后介绍 Flink 的关键技术，最初讲讲 Flink 在快手生产实践中的一些利用，包含实时指标计算和疾速 failover。

一、流式计算的介绍

流式计算次要针对 unbounded data（无界数据流）进行实时的计算，将计算结果疾速的输入或者修改。

这部分将分为三个大节来介绍。第一，介绍大数据系统发展史，包含初始的批处理到当初比拟成熟的流计算；第二，为大家简略比照下批处理和流解决的区别；第三，介绍流式计算外面的关键问题，这是每个优良的流式计算引擎所必须面临的问题。

1、大数据系统发展史

上图是 2003 年到 2018 年大数据系统的发展史，看看是怎么一步步走到流式计算的。

2003 年，Google 的 MapReduce 横空出世，通过经典的 Map&Reduce 定义和零碎容错等保障来不便解决各种大数据。很快就到了 Hadoop，被认为是开源版的  MapReduce，带动了整个 apache 开源社区的凋敝。再往后是谷歌的 Flume，通过算子连贯等 pipeline 的形式解决了多个 MapReduce 作业连贯解决低效的问题。

流式零碎的开始以 Storm 来介绍。Storm 在 2011 年呈现，具备延时短、性能低等个性，在过后颇受青睐。然而 Storm 没有提供零碎级别的 failover 机制，无奈保障数据一致性。那时的流式计算引擎是不准确的，lamda 架构组装了流解决的实时性和批处理的准确性，已经风行一时，起初因为难以保护也逐步败落。

接下来呈现的是 Spark Streaming，能够说是第一个生产级别的流式计算引擎。Spark Streaming 晚期的实现基于成熟的批处理，通过 mini batch 来实现流计算，在 failover 时可能保障数据的一致性。

Google 在流式计算方面有很多摸索，包含 MillWheel、Cloud Dataflow、Beam，提出了很多流式计算的理念，对其余的流式计算引擎影响很大。

再来看 Kafka。Kafka 并非流式计算引擎，然而对流式计算影响特地大。Kafka 基于 log 机制、通过 partition 来保留实时数据，同时也能存储很长时间的历史数据。流式计算引擎能够无缝地与 kafka 进行对接，一旦呈现 Failover，能够利用 Kafka 进行数据回溯，保证数据不失落。另外，Kafka 对 table 和 stream 的摸索特地多，对流式计算影响微小。

Flink 的呈现也比拟久，始终到 2016 年左右才火起来的。Flink 借鉴了很多 Google 的流式计算概念，使得它在市场上特地具备竞争力。前面我会具体介绍 Flink 的一些特点。

2、批处理与流计算的区别

批处理和流计算有什么样的区别，这是很多同学有疑难的中央。咱们晓得 MapReduce 是一个批处理引擎，Flink 是一个流解决引擎。咱们从四个方面来进行一下比照：

1）应用场景

MapReduce 是大批量文件解决，这些文件都是 bounded data，也就是说你晓得这个文件什么时候会完结。相比而言，Flink 解决的是实时的 unbounded data，数据源源不断，可能永远都不会完结，这就给数据齐备性和 failover 带来了很大的挑战。

2）容错

MapReduce 的容错伎俩包含数据落盘、反复读取、最终后果可见等。文件落盘能够无效保留两头后果，一旦 task 挂掉重启就能够间接读取磁盘数据，只有作业胜利运行完了，最终后果才对用户可见。这种设计的哲理就是你能够通过反复读取同一份数据来产生同样的后果，能够很好的解决 failover。

Flink 的容错次要通过定期快照和数据回溯。每隔一段时间，Flink 就会插入一些 barrier，barrier 从 source 流动到 sink，通过 barrier 流动来管制快照的生成。快照制作完就能够保留在共享引擎里。一旦作业呈现问题，就能够从上次快照进行复原，通过数据回溯来从新生产。

3）性能

MapReduce 次要特点是高吞吐、高延时。高吞吐阐明解决的数据量十分大；高延时就是后面说到的容错问题，它必须把整个作业处理完才对用户可见。

Flink 次要特点是高吞吐、低延时。在流式零碎里，Flink 的吞吐是很高的。同时，它也能够做到实时处理和输入，让用户疾速看到后果。

4）计算过程

MapReduce 次要通过 Map 和 reduce 来计算。Map 负责读取数据并作根本的解决，reduce 负责数据的聚合。用户能够依据这两种根本算子，组合出各种各样的计算逻辑。

Flink 为用户提供了 pipeline 的 API 和批流对立的 SQL。通过 pipeline  的 API，用户能够不便地组合各种算子构建简单的利用；Flink SQL 是一个更高层的 API 形象，极大地升高了用户的应用门槛。

3、流式计算的关键问题

这部分次要通过四个问题给大家解答流式计算的关键问题，也是很多计算引擎须要思考的问题。

1）What

What 是指通过什么样的算子来进行计算。次要蕴含三个方面的类型，element-wise 示意一对一的计算，aggregating 示意聚合操作，composite 示意多对多的计算。

2）Where

aggregating 会进行一些聚合的计算，次要是在各种 window 里进行计算。窗口蕴含滑动窗口、滚动窗口、会话窗口。窗口会把无界的数据切分成有界的一个个数据块进行解决，前面咱们会具体介绍这点。

3）When

When 就是什么时候触发计算。窗口外面有数据，因为输出数据是无穷无尽的，很难晓得一个窗口的数据是否全副达到了。流式计算次要通过 watermark 来保障数据的齐备性，通过 trigger 来决定何时触发。当接管到数值为 X 的 Watermark 时，能够认为所有工夫戳小于等于 X 的事件全副达到了。一旦 watermark 跨过窗口完结工夫，就能够通过 trigger 来触发计算并输入后果。

4）How

How 次要指咱们如何从新定义同一窗口的屡次触发后果。后面也说了 trigger 是用来触发窗口的，一个窗口可能会被触发屡次，比方 1 分钟的窗口每 10 秒触发计算一次。解决形式次要蕴含三种：

• Discarding，抛弃之前的状态从新计算。这种形式每次的触发后果都是互不关联的，屡次触发后果的组合反映了全副的窗口内容，上游个别会再次聚合；
• Accumulating，这个就是一个聚合的状态，比如说第二次触发的时候是在第一次的后果上进行计算的，上游只须要保留最新的后果即可；
• Accumulating 和 retracting，这个次要在 Accumulating 的根底上加了一个 retracting，retracting 的意思就是撤销。窗口再次触发时，会通知上游撤销上一次的计算结果，并告知最新的后果。Flink SQL 的聚合就应用了这种 retract 的模式。

二、Flink 关键技术

1、Flink 简介

Flink 是一款分布式计算引擎，既能够进行流式计算，也能够进行批处理。下图是官网对 Flink 的介绍：

Flink 能够运行在 k8s、yarn、mesos 等资源调度平台上，依赖 hdfs 等文件系统，输出蕴含事件和各种其余数据，通过 Flink 引擎计算后再输入到其余中间件或者数据库等。

Flink 有两个外围概念：

• State：Flink 能够解决有状态的数据，通过本身的 state 机制来保障作业 failover 时数据不失落；
• Event Time：容许用户依照事件工夫来解决数据，通过 watermark 来推动工夫后退，这个前面还会具体介绍。次要是零碎的工夫和事件的工夫。

Flink 次要通过下面两个核心技术来保障 exactly-once，比如说作业 Failover 的时候状态不失落，就如同没产生故障一样。

2、快照机制

Flink 的快照机制次要是为了保障作业 failover 时不失落状态。Flink 提供了一种轻量级的快照机制，不须要进行作业就能够帮忙用户长久化内存中的状态数据。

上图中的 markers（与 barrier 语义雷同）通过流动来触发快照的制作，每一个编号都代表了一次快照，比方编号为 n 的 markers 从最上游流动到最上游就代表了一次快照的制作过程。简述如下：

• 零碎发送编号为 n 的 markers 到最上游的算子，markers 随着数据往上游流动；
• 当上游算子收到 marker 后，就开始将本身的状态保留到共享存储中；
• 当所有最上游的算子接管到 marker 并实现算子快照后，本次作业的快照制作实现。

一旦作业失败，重启时就能够从快照复原。

上面为一个简略的 demo 阐明（barrier 等同于 marker）。

• barrier 达到 Source，将状态 offset=7 存储到共享存储；
• barrier 达到 Task，将状态 sum=21 存储到共享存储；
• barrier 达到 Sink，commit 本次快照，标记着快照的胜利制作。

这时候忽然间作业也挂掉，重启时 Flink 会通过快照复原各个状态。Source 会将本身的 offset 置为 7，Task 会将本身的 sum 置为 21。当初咱们能够认为 1、2、3、4、5、6 这 6 个数字的加和后果并没有失落。这个时候，offset 从 7 开始生产，跟作业失败前齐全对接了起来，确保了 exactly-once。

3、事件工夫

工夫类型分为两种：

• Event time（事件工夫），指事件产生的工夫，比方采集数据时的工夫；
• Processing time（零碎工夫），指零碎的工夫，比方解决数据时的工夫。

如果你对数据的准确性要求比拟高的话，采纳 Event time 能保障 exactly-once。Processing Time 个别用于实时生产、精准性要求略低的场景，次要是因为工夫生成不是 deterministic。

咱们能够看上面的关系图，X 轴是 Event time，Y 轴是 Processing time。现实状况下 Event time 和 Processing time 是雷同的，就是说只有有一个事件产生，就能够立即解决。然而理论场景中，事件产生后往往会通过肯定延时才会被解决，这样就会导致咱们零碎的工夫往往会滞后于事件工夫。这里它们两个的差 Processing-time lag 示意咱们处理事件的延时。

事件工夫罕用在窗口中，应用 watermark 来确保数据齐备性，比如说 watermarker 值大于 window 开端工夫时，咱们就能够认为 window 窗口所有数据都曾经达到了，就能够触发计算了。

比方下面 [0-10] 的窗口，当初 watermark 走到了 10，曾经达到了窗口的完结，触发计算 SUM=21。如果要是想对早退的数据再进行触发，能够再定义一下前面 late data 的触发，比如说前面来了个 9，咱们的 SUM 就等于 30。

4、窗口机制

窗口机制就是把无界的数据分成数据块来进行计算，次要有三种窗口。

• 滚动窗口：固定大小的窗口，相邻窗口没有交加；
• 滑动窗口：每个窗口的大小是一样的，然而两个窗口之间会有重合；
• 会话窗口：依据沉闷工夫聚合而成的窗口，比方沉闷工夫超过 3 分钟新起一个窗口。窗口之间留有肯定的距离。

窗口会主动治理状态和触发计算，Flink 提供了丰盛的窗口函数来进行计算。次要包含以下两种：

• ProcessWindowFunction，全量计算会把所有数据缓存到状态里，始终到窗口完结时对立计算。相对来说，状态会比拟大，计算效率也会低一些；
• AggregateFunction，增量计算就是来一条数据就算一条，可能咱们的状态就会特地的小，计算效率也会比 ProcessWindowFunction 高很多，然而如果状态存储在磁盘频繁拜访状态可能会影响性能。

三、快手 Flink 实际

1、利用概括

快手利用概括次要是分为数据接入、Flink 实时计算、数据利用、数据展现四个局部。各层各司其职、连接晦涩，为用户提供一体化的数据服务流程。

2、实时指标计算

常见的实时指标计算包含 uv、pv 和 sum。这其中 uv 的计算最为简单也最为经典。上面我将重点介绍 uv。

uv 指的是不同用户的个数，咱们这边计算的就是不同 deviceld 的个数，次要的挑战来自三方面：

• 用户数多，数据量大。流动期间的 QPS 常常在千万级别，理论计算起来特地简单；
• 实时性要求高，通常为几秒到分钟后果的输入；
• 稳定性要求高，比如说咱们在做春晚流动时候要求故障工夫须要低于 2% 或更少。

针对各种各样的 uv 计算，咱们提供了一套成熟的计算流程。次要蕴含了三方面：

• 字典计划：将 string 类型的 deviceld 转成 long 类型，不便后续的 uv 计算；
• 歪斜解决：比方某些大 V 会导致数据重大歪斜，这时候就须要打散解决；
• 增量计算：比方计算 1 天的 uv，每分钟输入一次后果。

字典计划须要确保任何两个不同的 deviceId 不能映射到雷同的 long 类型数字上。快手外部次要应用过以下三种计划：

• HBase，基于 partition 分区建设 deviceld 到 id 的映射，通过缓存和批量拜访来减速；
• Redis，这种计划严格来说不属于字典，次要通过 key-value 来判断数据是否首次呈现，基于首次数据来计算 uv，这样就会把 pv 和 uv 的计算进行对立;
• 最初就是一个 Flink 外部自建的全局字典实现 deviceld 到 id 的转换，之后计算 UV。

这三种计划外面，前两种属于内部存储的字典计划，长处是能够做到多个作业共享 1 份数据，毛病是内部拜访慢而且不太稳固。最初一种 Flink 字典计划基于 state，不依赖内部存储，性能高然而无奈多作业共享。

接下来咱们重点介绍基于 Flink 本身的字典计划，下图次要是建设一个 deviceld 到 id 的映射：

次要分成三步走：

1）建设 Partition 分区，指定一个比拟大的 Partition 分区个数，该个数比拟大并且不会变，依据 deviceld 的哈希值将其映射到指定 partition。

2）建设 id 映射。每个 Partition 都有本人负责的 id 区间，确保 Partition 之间的 long 类型的 id 不反复，partition 外部通过自增 id 来确保每个 deviceId 对应一个 id。

3）应用 keyed state 保留 id 映射。这样咱们的作业呈现并发的大扭转时，能够不便的 rescale，不须要做其余的操作。

除了 id 转换，前面就是一个实时指标计算的常见问题，就是数据歪斜。业界常见的解决数据歪斜解决计划次要是两种：

• 打散再聚合：先将歪斜的数据打散计算，而后再聚合计算结果；
• Local-aggregate：先在本地计算预聚合，这样会大大减少上游的数据压力。

二者的实质是一样的，都是先预聚合再汇总，从而防止单点性能问题。

上图为计算最小值的热点问题，红色数据为热点数据。如果间接将它们打到同一个分区，会呈现性能问题。为了解决歪斜问题，咱们通过 hash 策略将数据分成小的 partition 来计算，如上图的预计算，最初再将两头后果汇总计算。

当所有就绪后，咱们来做增量的 UV 计算，比方计算 1 天 uv，每分钟输入 1 次后果。计算形式既能够采纳 API，也能够采纳 SQL。

针对 API，咱们抉择了 global state+bitmap 的组合，既严格遵循了 Event Time 又缩小了 state 大小：

上面为计算流程（须要留神时区问题）：

• 定义跟触发距离一样大小的 window（比方 1 分钟）；
• Global state 用来保留跨窗口的状态，咱们采纳 bitmap 来存储状态；
• 每隔一个 window 触发一次，输入起始至今的 UV；
• 以后作用域（比方 1 天）完结，清空状态从新开始。

针对 SQL，增量计算反对的还不是那么欠缺，然而能够利用 early-fire 的参数来提前触发窗口。

配置如下：

table.exec.emit.early-fire.enabled：truetable.exec.emit.early-fire.delay：60 s

early-fire.delay 就是每分钟输入一次后果的意思。

SQL 如下：

SELECT TUMBLE_ROWTIME(eventTime, interval‘1’day) AS rowtime, dimension, count(distinct id) as uv
FROM person
GROUP BY TUMBLE(eventTime, interval '1' day), dimension

如果遇到歪斜，能够参考上一步来解决。

3、疾速 failover

最初看下咱们部门最近发力的一个方向，如何疾速 failover。

Flink 作业都是 long-running 的在线作业，很多对可用性的要求特地高，尤其是跟公司外围业务相干的作业，SLA 要求 4 个 9 甚至更高。当作业遇到故障时，如何疾速复原对咱们来说是一个微小的挑战。

上面分三个方面来开展：

• Flink 以后已有的疾速复原计划；
• 基于 container 宕掉的疾速复原；
• 基于机器宕掉的疾速复原。

1）Flink 以后已有的疾速复原计划

Flink 以后已有的疾速复原计划次要包含以下两种：

• region failover。如果流式作业的 DAG 蕴含多个子图或者 pipeline，那么 task 失败时只会影响其所属的子图或者 pipeline，而不必整个 DAG 都重新启动；
• local recovery。在 Flink 将快照同步到共享存储的同时，在本地磁盘也保留一份快照。作业失败复原时，能够调度到上次部署的地位，并从 local disk 进行快照复原。

2）基于 container 宕掉的疾速复原

理论环境中，container 宕掉再申请有时会长达几十秒，比方因为 hdfs 慢、yarn 慢等起因，重大影响复原速度。为此，咱们做了如下优化：

• 冗余资源。维持固定个数的冗余 container，一旦 container 宕掉，冗余 container 立即候补上来，省去了繁冗的资源申请流程；
• 提前申请。一旦发现作业因为 container 宕掉而失败，立即申请新的 container。

以上优化笼罩了很大一部分场景，复原工夫从 30s-60s 降到 20s 以内。

3）基于机器宕掉的疾速复原

机器宕掉时，flink on yarn 的复原工夫超过 3 分钟，这对重要作业显然是无奈容忍的！为了做到疾速复原，咱们须要做到疾速感知和复原：

• 冗余资源并打散调配，确保两个冗余资源不在一个 container，redundantContainerNum=max(containerNumOfHost) + 1；
• 作业宕机，Hawk 监测零碎 5 秒内发现；
• 冗余资源疾速候补，免去申请资源的流程。

通过这种计划，咱们能够容忍任意一台机器的宕机，并将宕机复原工夫由原先的 3 分钟升高到 30 秒以内。

四、总结

本文从大数据系统的倒退动手，进而延长出流式零碎的要害概念，之后介绍了 Flink 的要害个性，最初解说了快手外部的实时指标计算和疾速 failover，心愿对大家有所帮忙。

五、Q&A

Q1：打算做实时计算，能够跳过 Storm、Spark 间接上手 Flink 吗？

A：能够间接应用 Flink。Storm 在 failover 时会失落数据，无奈做到 exactly-once；spark streaming 是 Flink 的竞争者，是在批处理的根底上实现流计算，相比而言，Flink 的底层是流解决，更加适宜流计算。

Q2：个别怎么解决 taskmanager heartbeat timeout？

A：默认 10 秒汇报一次心跳，心跳超时为 50 秒，这个时候作业会失败，如果配置了高可用那么会重启。

Q3：如何保障 2 天大时间跨度提早音讯的窗口计算？

A：这里次要的挑战在于工夫长、状态大，倡议 stateBakend 应用 Rocksdb（能够利用磁盘存储大状态），窗口计算倡议应用增量计算来缩小状态的大小。

Q4：Flink on Yarn，Yarn 重启会主动拉起 Flink 工作吗，说不能拉起怎么解决，手动启动吗？

A：如果配置了高可用（依赖 zookeeper），作业失败了就能够主动拉起。

Q5：Kafka 目前多用作数据直达平台，Flink 相当于代替了 Kafka Stream 吗？

 A：Kafka 的外围性能是消息中间件，kafka stream 能够跟 kafka 很好的集成，但并不是一个业余的计算引擎。相比而言，flink 是一个分布式的流式计算引擎，性能上更加弱小。

Q6：你们怎么对待 Apache Beam？

A：Apache Beam 在下层进行了形象，能够类比 SQL，只定义标准，底层能够接入各种计算引擎。