乐趣区

关于Flink:Flink-执行引擎流批一体的融合之路

jiezi

本文由 Apache Flink Committer 马国维分享，次要介绍 Flink 作为大数据计算引擎的流批一体交融之路。内容包含：

1、背景
2、流批一体的分层架构
3、流批一体 DataStream
4、流批一体 DAG Scheduler
5、流批一体的 Shuffle 架构
6、流批一体的容错策略
7、将来瞻望

一、背景

随着互联网和挪动互联网的一直倒退，各行各业都积攒海量的业务数据。而企业为了改善用户体验，晋升产品在市场上的竞争力，都采取了实时化形式来解决大数据。社交媒体的实时大屏、电商的实时举荐、城市大脑的实时交通预测、金融行业的实时反欺诈，这些产品的胜利都在阐明大数据处理的实时化曾经成为一个势不可挡的潮流。

在实时化的大趋势下，Flink 曾经成为实时计算行业的事实标准。咱们看到，不光是阿里巴巴，国内外各个领域的头部厂商，都把 Flink 做为实时计算的技术底座，国内有字节跳动、腾讯、华为，国外有 Netflix、Uber 等等。

而业务实时化只是一个终点，Flink 的指标之一就是给用户提供实时离线一体化的用户体验。其实很多用户不仅须要实时的数据统计，为了确认经营或产品的策略的成果，用户同时还须要和历史（昨天，甚至是去年的同期）数据比拟。而从用户的角度来看，原有的流、批独立计划存在一些痛点：

  • 人力老本比拟高。因为流和批是两套零碎，雷同的逻辑须要两个团队开发两遍。
  • 数据链路冗余。在很多的场景下，流和批计算内容其实是统一，然而因为是两套零碎，所以雷同逻辑还是须要运行两遍，产生肯定的资源节约。
  • 数据口径不统一。这个是用户遇到的最重要的问题。两套零碎、两套算子，两套 UDF，肯定会产生不同水平的误差，这些误差给业务方带来了十分大的困扰。这些误差不是简略依附人力或者资源的投入就能够解决的。

2020 年的双十一，在实时洪峰达到 40 亿的历史新高的同时，Flink 团队与 DT 团队一起推出了基于 Flink 的全链路流批一体的数仓架构，很好解决了 Lambda 的架构所带来的一系列问题：流批作业应用同一 SQL，使研发效率晋升了 3~4 倍；一套引擎确保了数据口径人造统一；流批作业在同一集群运行，削峰填谷大幅晋升了资源效率。

Flink 流批一体的胜利，离不开 Flink 开源社区的衰弱蓬勃发展。从 Apache 软件基金会 2020 年度报告能够看出，在反映开源社区凋敝状况的三个要害指标上 Flink 都名落孙山：用户邮件列表活跃度，Flink 排名第一；开发者提交次数 Flink 排名第二，Github 用户访问量排名第二。这些数据并不局限于大数据畛域，而是 Apache 开源基金会上司的所有我的项目。

2020 年也是 Blink 反哺社区的第二年，这两年咱们把 Blink 在团体内积攒的教训逐渐奉献回社区，让 Flink 成为真正意义上的流批一体平台。我心愿通过这篇文章给大家分享下这两年 Flink 在执行引擎流批交融方面做了哪些工作。同时也心愿 Flink 的老用户和新敌人能够进一步理解 Flink 流批一体架构的“前世今生”。

二、流批一体的分层架构

总体来说，Flink 的外围引擎次要分为如下三层：

  • SDK 层。Flink 的 SDK 次要有两类，第一类是关系型 Relational SDK 也就是 SQL/Table，第二类是物理型 Physical SDK 也就是 DataStream。这两类 SDK 都是流批对立，即不论是 SQL 还是 DataStream，用户的业务逻辑只有开发一遍，就能够同时在流和批的两种场景下应用；
  • 执行引擎层。执行引擎提供了对立的 DAG，用来形容数据处理流程 Data Processing Pipeline(Logical Plan)。不论是流工作还是批工作，用户的业务逻辑在执行前，都会先转化为此 DAG 图。执行引擎通过 Unified DAG Scheduler 把这个逻辑 DAG 转化成在分布式环境下执行的 Task。Task 之间通过 Shuffle 传输数据，咱们通过 Pluggable Unified Shuffle 架构，同时反对流批两种 Shuffle 形式；
  • 状态存储。状态存储层负责存储算子的状态执行状态。针对流作业有开源 RocksdbStatebackend、MemoryStatebackend，也有商业化的版本的 GemniStateBackend；针对批作业咱们在社区版本引入了 BatchStateBackend。

本文次要分享如下几个方面的内容：

  1. 流批一体的 DataStream 介绍了如何通过流批一体的 DataStream 来解决 Flink SDK 以后面临的挑战；
  2. 流批一体的 DAG Scheduler 介绍了如何通过对立的 Pipeline Region 机制充沛开掘流式引擎的性能劣势；如何通过动静调整执行打算的形式来改善引擎的易用性，进步零碎的资源利用率；
  3. 流批一体的 Shuffle 架构介绍如何通过一套对立的 Shuffle 架构既能够满足不同 Shuffle 在策略上的定制化需要，同时还能防止在共性需要上的反复开发；
  4. 流批一体的容错策略介绍了如何通过对立的容错策略既满足批场景下容错又能够晋升流场景下的容错成果。

三、流批一体 DataStream

SDK 剖析以及面临的挑战

如上图所示，目前 Flink 提供的 SDK 次要有三类：

  1. Table/SQL 是一种 Relational 的高级 SDK，次要用在一些数据分析的场景中，既能够反对 Bounded 也能够反对 Unbounded 的输出。因为 Table/SQL 是 Declarative 的，所以零碎能够帮忙用户进行很多优化，例如依据用户提供的 Schema，能够进行 Filter Push Down 谓词下推、按需反序列二进制数据等优化。目前 Table/SQL 能够反对 Batch 和 Streaming 两种执行模式。[1]
  2. DataStream 属于一种 Physical SDK。Relatinal SDK 性能尽管弱小，但也存在一些局限：不反对对 State、Timer 的操作；因为 Optimizer 的降级，可能导致用雷同的 SQL 在两个版本中呈现物理执行打算不兼容的状况。而 DataStream SDK，既能够反对 State、Timer 维度 Low Level 的操作，同时因为 DataStream 是一种 Imperative SDK，所以对物理执行打算有很好的“掌控力”，从而也不存在版本升级导致的不兼容。DataStream 目前在社区仍有很大用户群，例如目前未 Closed 的 DataStream issue 仍然有近 500 个左右。尽管 DataStream 即能够反对 Bounded 又能够反对 Unbounded Input 用 DataStream 写的 Application，然而在 Flink-1.12 之前只反对 Streaming 的执行模式。
  3. DataSet 是一种仅反对 Bounded 输出的 Physical SDK，会依据 Bounded 的个性对某些算子进行做肯定的优化，然而不反对 EventTime 和 State 等操作。尽管 DataSet 是 Flink 提供最早的一种 SDK，然而随着实时化和数据分析场景的一直倒退，相比于 DataStream 和 SQL，DataSet 在社区的影响力在逐渐降落。

目前 Table/SQL 对于流批对立的场景反对曾经比拟成熟，然而对于 Phyiscal SDK 来说还面临的一些挑战，次要有两个方面：

  1. 利用已有 Physical SDK 无奈写出一个真正生产能够用的流批一体的 Application。例如用户写一个程序用来解决 Kafka 中的实时数据，那么利用雷同的程序来解决存储在 OSS/S3/HDFS 上的历史数据也是十分天然的事件。然而目前不论是 DataSet 还是 DataStream 都无奈满足用户这个“简略”的诉求。大家可能感觉奇怪，DataStream 不是既反对 Bounded 的 Input 又反对 Unbounded 的 Input，为什么还会有问题呢？其实“魔鬼藏在细节中”，我会在 Unified DataStream 这一节中会做进一的论述。
  2. 学习和了解的老本比拟高。随着 Flink 一直壮大，越来越多的新用户退出 Flink 社区，然而对于这些新用户来说就要学习两种 Physical SDK。和其余引擎相比，用户入门的学习老本是绝对比拟高的；两种 SDK 在语义上有不同的中央，例如 DataStream 上有 Watermark、EventTime，而 DataSet 却没有，对于用户来说，了解两套机制的门槛也不小；因为这两 SDK 还不兼容，一个新用户一旦抉择谬误，将会面临很大的切换老本。

Unified Physical SDK

为了解决上述 Physical SDK 所面临的挑战，咱们把 Unified DataStream SDK 作为 Flink 对立的 Physical SDK。这个局部次要解决两个问题：

  1. 为什么抉择 DataStream 作为 Unified Physical SDK？
  2. Unified DataStream 比“老”的 DataStream 提供了哪些能力让用户能够写出一个真正生产能够用的流批一体 Application？

为什么不是 Unified DataSet

为了解决学习和了解老本比拟高的问题，最天然最简略的计划就是从 DataStream 和 DataSet 中抉择一个作为 Flink 的惟一的 Physical SDK。那为什么咱们抉择了 DataStream 而不是 DataSet 呢？次要有两个起因：

  1. 用户收益。在前边曾经剖析过，随着 Flink 社区的倒退，目前 DataSet 在社区的影响力逐步降落。如果抉择应用 DataSet 作为 Unified Physical SDK，那么用户之前在 DataStream 大量“投资”就会作废。而抉择 DataStream，能够让许多用户的已有 DataStream“投资”失去额定的回报；
  2. 开发成本。DataSet 过于古老，不足大量对于古代实时计算引擎基本概念的反对，例如 EventTime、Watermark、State、Unbounded Source 等。另外一个更深层的起因是现有 DataSet 算子的实现，在流的场景齐全无奈复用，例如 Join 等。而对于 DataStream 则不然，能够进行大量的复用。那么如何在流批两种场景下复用 DataStream 的算子呢？

Unified DataStream

很多对 Flink 有肯定理解的用户可能会问：DataStream 是同时反对 Bounded/Unbounded 的输出，为什么咱们会说：用 DataStream 无奈写出一个真正生产能够用的流批一体 Application 呢？简略来说，DataStream 本来次要设计指标是给 Unbounded 场景应用的，所以导致在 Bounded 的场景下在效率、可用性、易用性方面和传统的批引擎还有肯定间隔。具体来说体现在如下两个方面：

– 效率

先给大家看一个例子，下边是一个跑同样规模的 WordCount 的作业，DataStream 和 DataSet 的性能比照图。从这个例子能够看出，DataSet 的性能是 DataStream 将近 5 倍。

很显著，要让 DataStream 在生产中既能够反对流的场景又要反对批的场景，就肯定要大幅提高 DataStream 在 Bounded 场景下效率。那么为什么 DataStream 的效率要比 DataSet 的效率低呢？

后面咱们曾经提到，DataStream 本来次要设计指标是给 Unbounded 的场景下应用的，而 Unounded 场景下一个次要的特点就是乱序，也就是说任何一个 DataStream 的算子无奈假如解决的 Record 是依照什么程序进行的，所以许多算子会用一个 K/V 存储来缓存这些乱序的数据，等到适合的时候再从 K/V 存储中取出这些数据进行解决并且进行输入。个别状况下，算子对 K/V 存储拜访波及大量的序列化和反序列化，同时也会引发随机磁盘 I/O；而在 DataSet 中，假如数据是有界的，也就是能够通过优化来防止随机的磁盘 I/O 拜访，同时也对序列化和反序列化做了相干优化。这也是为什么用 DataSet 写的 WorkerCount 要比用 DataStream 写的 WordCount 快 5 倍次要起因。

晓得到了起因，是不是要把所有的 DataStream 的算子，都重写一遍就能够了呢？实践上没问题，然而 DataStream 有大量的算子须要重写，有些算子还比较复杂，例如与 Window 相干的一系列算子。能够设想到，如果都全副重写，工程量是十分之微小的。所以咱们通过单 Key 的 BatchStateBackend 简直完全避免了对所有算子重写，同时还失去了十分不错的成果。

– 一致性

对于 Flink 有肯定理解的同学应该都晓得，原来用 DataStream 写的 Application 都采纳 Streaming 的执行模式，在这种模式下是通过 Checkpoint 的形式放弃端到端的 Exactly Once 的语义，具体来说一个作业的 Sink 只有当全图的所有算子 (包含 Sink 本人) 都做完各自的 Snapshot 之后，Sink 才会把数据 Commit 到内部零碎中，这是一个典型的依赖 Flink Checkpoint 机制的 2PC 协定。

而在 Bounded 的场景下尽管也能够采纳 Streaming 的形式然而对于用户来说可能会存在一些问题：

  1. 资源耗费大: 应用 Streaming 形式，须要同时拿到所有的资源。在某些状况下，用户可能没有这么多资源；
  2. 容错老本高: 在 Bounded 场景下，为了效率一些算子可能无奈反对 Snapshot 操作，一旦出错可能须要从新执行整个作业。

所以在 Bounded 场景下，用户心愿 Application 能够采纳 Batch 执行模式，因为利用 Batch 执行的模式能够十分天然的解决上述两个问题。在 Bounded 场景下反对 Batch 的执行模式是比较简单的，然而却引入了一个十分辣手的问题——利用已有的 Sink API 无奈保障端到端的 Exactly Once 语义。这是因为 Bounded 场景下是没有 Checkpoint 的，而原有 Sink 都是依赖 Checkpoint 保障端到端的 ExactlyOnce 的。同时咱们不心愿开发者针对 Sink 在不同模式下开发两套不同的实现，因为这样十分不利用 Flink 和其余生态的对接。

实际上，一个 Transactional 的 Sink 次要解决如下 4 个问题：

  1. What to commit?
  2. How to commit?
  3. Where to commit?
  4. When to commit?

而 Flink 应该让 Sink 开发者提供 What to commit 和 How to commit，而零碎应该依据不同的执行模式，抉择 Where to commit 和 When to commit 来保障端到端的 Exactly Once。最终咱们提出了一个全新 Unified Sink API，从而让开发者只开发一套 Sink 就能够同时运行在 Streaming 和 Batch 执行模式下。这里介绍的只是次要思路，在无限流的场景下如何保障 End to End 的一致性；如何对接 Hive、Iceberg 等内部生态，实际上还是存在肯定挑战。

四、流批一体 DAG Scheduler

Unified DAG Scheduler 要解决什么问题

原来 Flink 有两种调度的模式：

  1. 一种是 流的调度模式，在这种模式下，Scheduler 会申请到一个作业所须要的全副资源，而后同时调度这个作业的全副 Task，所有的 Task 之间采取 Pipeline 的形式进行通信。批作业也能够采取这种形式，并且在性能上也会有很大的晋升。然而对于运行比拟长的 Batch 作业来说来说，这种模式还是存在肯定的问题：规模比拟大的状况下，同时耗费的资源比拟多，对于某些用户来说，他可能没有这么多的资源；容错代价比拟高，例如一旦产生谬误，整个作业都须要从新运行。
  2. 一种是 批的调度模式。这种模式和传统的批引擎相似，所有 Task 都是能够独立申请资源，Task 之间都是通过 Batch Shuffle 进行通信。这种形式的益处是容错代价比拟小。然而这种运行形式也存在一些短板。例如，Task 之间的数据都是通过磁盘来进行交互，引发了大量的磁盘 IO。

总的来说，有了这两种调度形式是能够根本满足流批一体的场景需要，然而也存在着很大的改良空间，具体来说体现在三个方面：

1、架构不统一、保护老本高 。调度的实质就是进行资源的调配，换句话说就是要解决 When to deploy which tasks to where 的问题。原有两种调度模式，在资源分配的机会和粒度上都有肯定的差别，最终导致了调度架构上无奈齐全对立，须要开发人员保护两套逻辑。例如，流的调度模式，资源分配的粒度是整个物理执行打算的全副 Task；批的调度模式，资源分配的粒度是单个工作，当 Scheduler 拿到一个资源的时候，就须要依据作业类型走两套不同的解决逻辑；
2、 性能 。传统的批调度形式，尽管容错代价比拟小，然而引入大量的磁盘 I/O，并且性能也不是最佳，无奈施展出 Flink 流式引擎的劣势。实际上在资源绝对短缺的场景下，能够采取“流”的调度形式来运行 Batch 作业，从而防止额定的磁盘 I/O，进步作业的执行效率。尤其是在夜间，流作业能够开释出肯定资源，这就为批作业依照“Streaming”的形式运行提供了可能。
3、 自适应。目前两种调度形式的物理执行打算是动态的，动态生成物理执行打算存在调优人力老本高、资源利用率低等问题。

基于 Pipeline Region 的对立调度

为了既能施展流引擎的劣势，同时防止全图同时调度存在的一些短板，咱们引入 Pipeline Region 的概念。Unified DAG Scheduler 容许在一个 DAG 图中，Task 之间既能够通过 Pipeline 通信，也能够通过 Blocking 形式进行通信。这些由 Pipeline 的数据交换形式连贯的 Task 被称为一个 Pipeline Region。基于以上概念，Flink 引入 Pipeline Region 的概念，不论是流作业还是批作业，都是依照 Pipeline Region 粒度来申请资源和调度工作。仔细的读者能够发现，其实原有的两种模式都是 Pipeline Region 调度的特例。

即使能够资源上满足“流”的调度模式，那么哪些工作能够采取“流”的形式调度呢？

有同学还是会放心采取“流”的调度形式容错代价会比拟高，因为在“流”的调度形式下，一个 Task 产生谬误，和他联通的所有 Task 都会 Fail，而后从新运行。

在 Flink 中，不同 Task 之间有两种连贯形式[2]，一种是 All-to-All 的连贯形式，上游 Task 会和上游的所有的 Task 进行连贯；一种是 PointWise 的链接形式，上游的 Task 只会和上游的局部 Task 进行连贯。

如果一个作业的所有 Task 之间都是通过 All-to-All 形式进行连贯，一旦采取“流”的调度形式，那么整个物理拓扑都须要同时被调度，那么的确存在 FailOver 代价比拟高的问题[3]。然而在理论 Batch 作业的拓扑中，Task 之间不都是通过 All-to-All 的边连贯，Batch 作业中存在的大量 Task 通过 PointWise 的边连贯，通过“流”的形式调度由 PointWise 连贯的 Task 连通图，在缩小作业的容错老本的同时，能够进步作业的执行效率，如下图所示在，在全量的 10T TPC-DS 测试中，开启所有 PointWise 边都采纳 Pipeline 的链接形式就能够让整性能有 20% 以上的性能晋升。

上述只是 Schduler 提供的划分 Pipeline Region 的 4 种策略中的一种[4]，实际上 Planner 能够依据理论运行场景，定制哪些 Task 之间采取 Pipeline 的传输方式，哪些 Task 之间采取 Batch 的传输方式形式。

自适应调度

调度的实质是给物理执行打算进行资源分配的决策过程。Pipeline Region 解决了物理执行打算确定之后，流作业和批作业能够对立依照 Pipeline Region 的粒度进行调度。对于批作业来说动态生成物理执行打算存在一些问题[5]：

  1. 配置人力老本高。对于批作业来说，尽管实践上能够依据统计信息推断出物理执行打算中每个阶段的并发度，然而因为存在大量的 UDF 或者统计信息的缺失等问题，导致动态决策后果可能会呈现重大不精确的状况；为了保障业务作业的 SLA，在大促期间，业务的同学须要依据大促的流量预计，手动调整高优批作业的并发度，因为业务变动快，一旦业务逻辑发生变化，又要一直的反复这个过程。整个调优过程都须要业务的同学手动操作，人力老本比拟高，即使这样也可能会呈现误判的状况导致无奈满足用户 SLA;
  2. 资源利用率低。因为人工配置并发度老本比拟高，所以不可能对所有的作业都手动配置并发度。对于中低优先级的作业，业务同学会选取一些默认值作为并发度，然而在大多数状况下这些默认值都偏大，造成资源的节约；而且尽管高优先级的作业能够进行手工并发配置，因为配置形式比拟繁琐，所以大促过后，尽管流量曾经降落然而业务方依然会应用大促期间的配置，也造成大量的资源节约景象;
  3. 稳定性差。资源节约的状况最终导致资源的超额申请景象。目前大多数批作业都是采取和流作业集群混跑的形式，具体来说申请的资源都是非保障资源，一旦资源缓和或者呈现机器热点，这些非保障资源都是优先被调整的对象。

为了解决动态生成物理执行存在这些问题，咱们为批作业引入了自适应调度性能[6]，和原来的动态物理执行打算相比，利用这个个性能够大幅提高用户资源利用率。Adaptive Scheduler 能够依据一个 JobVertex 的上游 JobVertex 的执行状况，动静决定以后 JobVertex 的并发度。在将来，咱们也能够依据上游 JobVertex 产出的数据，动静决定上游采纳什么样的算子。

五、流批一体的 Shuffle 架构

Flink 是一个流批一体的平台，因而引擎对于不同的执行模式要别离提供 Streaming 和 Batch 两种类型的 Shuffle。尽管 Streaming Shuffle 和 Batch Shuffle 在具体的策略上存在肯定的差别，然而实质上都是为了对数据进行从新划分(re-partition)，因而不同的 Shuffle 之间还存在肯定的共性。所以咱们的指标是提供一套对立的 Shuffle 架构，既能够满足不同 Shuffle 在策略上的定制，同时还能防止在共性需要上进行反复开发。

总体来说，Shuffle 架构能够划分成如下图所示的四个档次。流和批的 Shuffle 需要在各层上有肯定差别，也有大量的共性，下边我做了一些简要剖析。

流批 Shuffle 之间的差别

大家都晓得，批作业和流作业对 Shuffle 的需要是有差别的，具体能够体现在如下 3 个方面:

1、Shuffle 数据的生命周期 。流作业的 Shuffle 数据和 Task 的生命周期根本是统一的；而批作业的 Shuffle 数据和 Task 生命周期是解耦的；
2、Shuffle 数据的存储介质。因为流作业的 Shuffle 数据生命周期比拟短，所以能够把流作业的 Shuffle 数据存储在内存中；而批作业的 Shuffle 数据生命周期有肯定的不确定性，所以须要把批作业的 Shuffle 数据存储在磁盘中；
3、Shuffle 部署形式[7]。 把 Shuffle 服务和计算节点部署在一起，对流作业来说这种部署形式是有劣势的，因为这样会缩小不必要网络开销，从而缩小 Latency。但对于批作业来说，这种部署形式在资源利用率、性能、稳定性上都存在肯定的问题。[8]

流批 Shuffle 之间的共性

批作业和流作业的 Shuffle 有差别也有共性，共性次要体现在：

1、数据的 Meta 治理 。所谓 Shuffle Meta 是指逻辑数据划分到数据物理地位的映射。不论是流还是批的场景，在失常状况下都须要从 Meta 中找出本人的读取或者写入数据的物理地位；在异常情况下，为了缩小容错代价，通常也会对 Shuffle Meta 数据进行长久化；
2、 数据传输。从逻辑上讲，流作业和批作业的 Shuffle 都是为了对数据进行从新划分(re-partition/re-distribution)。在分布式系统中，对数据的从新划分都波及到跨线程、过程、机器的数据传输。

流批一体的 Shuffle 架构

Unified Shuffle 架构形象出三个组件[9]: Shuffle Master、Shuffle Reader、Shuffle Writer。Flink 通过和这三个组件交互实现算子间的数据的从新划分。通过这三个组件能够满足不同 Shuffle 插件在具体策略上的差别：

  1. Shuffle Master 资源申请和资源开释。也就是说插件须要告诉框架 How to request/release resource。而由 Flink 来决定 When to call it;
  2. Shuffle Writer 上游的算子利用 Writer 把数据写入 Shuffle Service——Streaming Shuffle 会把数据写入内存；External/Remote Batch Shuffle 能够把数据写入到内部存储中;
  3. Shuffle Reader 上游的算子能够通过 Reader 读取 Shuffle 数据；

同时，咱们也为流批 Shuffle 的共性——Meta 治理、数据传输、服务部署[10]——提供了架构层面的反对，从而防止对简单组件的反复开发。高效稳固的数据传输，是分布式系统最简单的子系统之一，例如在传输中都要解决上下游反压、数据压缩、内存零拷贝等问题，在新的架构中只有开发一遍，就能够同时在流和批两种场景下独特应用，大大减少了开发和保护的老本。

六、流批一体的容错策略

Flink 原有容错策略是以检查点为前提的，具体来说无论是单个 Task 呈现失败还是 JobMaster 失败，Flink 都会依照最近的检查点重启整个作业。尽管这种策略存在肯定的优化空间，然而总的来说对于流的场景是根本是承受的。目前，Flink Batch 运行模式下不会开启检查点[11]，这也象征一旦呈现任何谬误，整个作业都要从头执行。

尽管原有策略在实践上能够保障最终肯定会产出正确的后果，然而显著大多数客户都无奈承受这种容错策略所付出的代价。为了解决这些问题，咱们别离对 Task 和 JM 的容错都做了相应的改良。

Pipeline Region Failover

尽管在 Batch 执行模式下没有定时的 Checkpoint，然而在 Batch 执行模式下，Flink 容许 Task 之间通过 Blocking Shuffle 进行通信。对于读取 Blocking Shuffle 的 Task 产生失败之后，因为 Blocking Shuffle 中存储了这个 Task 所须要的全副数据，所以只须要重启这个 Task 以及通过 Pipeline Shuffle 与其相连的全副上游工作即可，而不须要重启整个作业。

总的来说，Pipeline Region Failover 策略和 Scheduler 在进行失常调度的时候一样，都是把一个 DAG 拆分成由若干 Pipeline shuffle 连贯的一些 Pipeline Region，每当一个 Task 产生 FailOver 的时候，只会重启这个 Task 所在的 Region 即可。

JM Failover

JM 是一个作业的控制中心，蕴含了作业的各种执行状态。Flink 利用这些状态对工作进行调度和部署。一旦 JM 产生谬误之后，这些状态将会全副失落。如果没有这些信息，即使所有的工作节点都没有产生故障，新 JM 依然无奈持续调度原来的作业。例如，因为工作的完结信息都已失落，一个工作完结之后，新 JM 无奈判断现有的状态是否满足调度上游工作的条件——所有的输出数据都曾经产生。

从上边的剖析能够看出，JM Failover 的要害就是如何让一个 JM“复原记忆”。在 VVR[12] 中咱们通过基于 Operation Log 机制复原 JM 的要害状态。

仔细的同学可能曾经发现了，尽管这两个改良的出发点是为了批的场景，然而实际上对于流的作业容也同样无效。上边只是简要的介绍了两种容错策略的思路，实际上还有很多值得思考的内容。例如 Blocking 上游数据失落了咱们应该如何解决？JM 中有哪些要害的状态须要复原？

七、将来瞻望

为了提供比当初更快、更稳的用户体验，咱们曾经开始了下一代流式架构的研发；Flink 在流批一体的场景下失去了越来越多用户的认可，然而咱们也晓得业界还有很多高水平传统大数据系统值得咱们学习。最初我也心愿感兴趣的小伙伴能够退出咱们，一起打造一个具备完满用户体验的流批一体大数据计算引擎。

正文：

[1] Streaming 和 Batch 是两种执行模式和语义无关，Streaming 执行模式能够简略的了解为，Task 之间采纳 Pipeline 的 Shuffle；Batch 执行模式能够简略的了解为，Task 之间采纳 Blocking 的 Shuffle 模式。
[2] https://ci.apache.org/project…
[3] 咱们正在开发 Adaptive 的 Shuffle 模式，利用这种模式能够防止”纯”Pipeline 的形式引发的容错代价过高的问题。
[4] https://ci.apache.org/project…
[5] 对于流作业来说，动态物理执行打算也有和批相似的问题，咱们提供了一个 AutoPilot 的零碎来动静批改物理执行打算。因为 AutoPilot 属于独立的服务，不属于执行引擎这里就不开展赘述了。
[6] 因为工夫布局的起因，这个性能临时只存在咱们的商业化版本的执行引擎 VVR 中
[7] 在一些状况下，批 Shuffle Service 也会和计算节点部署在一起。例如，在 Flink Session 的模式下，尽管 Shuffle Service 和计算部署在一起有肯定的稳定性代价，然而对局部用户来说这种部署模式是在老本和稳定性之间衡量之下的一个后果。所以在肯定水平上，流批 Shuffle 在部署方面也是有共性的，只是不是完全相同而已。
[8] 把批作业的计算和 Shuffle 部署在一个节点内所存在的问题：资源利用率低、老本高。如果没有计算工作持续部署该节点上那么这个节点上计算资源就会被节约掉，计算资源提前开释也会节俭用户老本；性能无奈达到最优 因为一个节点只能看到局部的 Shuffle 数据，因而一个 Reduce 须要从 n 个节点上拉取本人的数据，这会引发大量的随机 IO 读，这样大量的随机读 IO 会大大降低作业性能；稳定性 一旦结点挂掉，整个节点所负责的 Shuffle 数据就会失落，而后就会触发作业从新计算，这种从新计算的代价都是比拟高的。(Task 中蕴含用户代码，所以此种结点 down 掉的概率会大于存储计算拆散状况下的 Shuffle 结点。)
[9] 因为历史起因，大家在读 Flink 代码的时候看到不是 Reader 和 Writer，而是 ResultPartion/InputGate。这里用 Reader 和 Writer 是为了升高刚接触 Flink 的同学的了解门槛。
[10] 部署为什么也算共性，能够参考[7]。
[11] 尽管实践上批作业能够反对检查点，然而在批的场景下，开启原生的流式 Checkpoint 老本是比拟高的的。当然这也不是齐全排除将来可能会发现比拟适合场景；
[12] VVR 是 Flink 商业产品的执行引擎，因为工夫布局的起因，这个性能临时还没有回馈给 Flink 社区

退出移动版