作者：王峰
整理：韩非
本文主要整理自云栖大会阿里巴巴计算平台事业部资深技术专家王峰（花名：莫问）在云栖大会‘开发者生态峰会’上发表的演讲。
伴随着海量增长的数据，数字化时代的未来感扑面而至。不论是结绳记事的小数据时代，还是我们正在经历的大数据时代，计算的边界正在被无限拓宽，而数据的价值，再也难以被计算。时下，谈及大数据，不得不提到最热门的下一代大数据计算引擎 Apache Flink（以下简称 Flink）。本文将结合 Flink 的前世今生，从业务角度出发，向大家娓娓道来：为什么阿里选择了 Flink？
合抱之木，生于毫末
随着人工智能时代的降临，数据量的爆发，在典型的大数据的业务场景下数据业务最通用的做法是：选用批处理的技术处理全量数据，采用流式计算处理实时增量数据。在绝大多数的业务场景之下，用户的业务逻辑在批处理和流处理之中往往是相同的。但是，用户用于批处理和流处理的两套计算引擎是不同的。因此，用户通常需要写两套代码。毫无疑问，这带来了一些额外的负担和成本。阿里巴巴的商品数据处理就经常需要面对增量和全量两套不同的业务流程问题，所以阿里就在想，我们能不能有一套统一的大数据引擎技术，用户只需要根据自己的业务逻辑开发一套代码。这样在各种不同的场景下，不管是全量数据还是增量数据，亦或者实时处理，一套方案即可全部支持，这就是阿里选择 Flink 的背景和初衷。
目前开源大数据计算引擎有很多选择，流计算如 Storm、Samza、Flink、Kafka Stream 等，批处理如 Spark、Hive、Pig、Flink 等。而同时支持流处理和批处理的计算引擎，只有两种选择: 一个是 Apache Spark, 一个是 Apache Flink。
从技术，生态等各方面的综合考虑，首先，Spark 的技术理念是基于批来模拟流的计算。而 Flink 则完全相反，它采用的是基于流计算来模拟批计算。
从技术发展方向看，用批来模拟流有一定的技术局限性，并且这个局限性可能很难突破。而 Flink 基于流来模拟批，在技术上有更好的扩展性。从长远来看，阿里决定用 Flink 做一个统一的、通用的大数据引擎作为未来的选型。
Flink 是一个低延迟、高吞吐、统一的大数据计算引擎。在阿里巴巴的生产环境中，Flink 的计算平台可以实现毫秒级的延迟情况下，每秒钟处理上亿次的消息或者事件。同时 Flink 提供了一个 Exactly-once 的一致性语义。保证了数据的正确性。这样就使得 Flink 大数据引擎可以提供金融级的数据处理能力。
Flink 在阿里的现状
基于 Apache Flink 在阿里巴巴搭建的平台于 2016 年正式上线，并从阿里巴巴的搜索和推荐这两大场景开始实现。目前阿里巴巴所有的业务，包括阿里巴巴所有子公司都采用了基于 Flink 搭建的实时计算平台。同时 Flink 计算平台运行在开源的 Hadoop 集群之上。采用 Hadoop 的 YARN 做为资源管理调度，以 HDFS 作为数据存储。因此，Flink 可以和开源大数据软件 Hadoop 无缝对接。
目前，这套基于 Flink 搭建的实时计算平台不仅服务于阿里巴巴集团内部，而且通过阿里云的云产品 API 向整个开发者生态提供基于 Flink 的云产品支持。
Flink 在阿里巴巴的大规模应用，表现如何？
- 规模：一个系统是否成熟，规模是重要指标，Flink 最初上线阿里巴巴只有数百台服务器，目前规模已达上万台，此等规模在全球范围内也是屈指可数；
- 状态数据：基于 Flink，内部积累起来的状态数据已经是 PB 级别规模；
- Events: 如今每天在 Flink 的计算平台上，处理的数据已经超过万亿条;
- TPS: 在峰值期间可以承担每秒超过 4.72 亿次的访问，最典型的应用场景是阿里巴巴双 11 大屏；
Flink 的发展之路
接下来从开源技术的角度，来谈一谈 Apache Flink 是如何诞生的，它是如何成长的？以及在成长的这个关键的时间点阿里是如何进入的？并对它做出了那些贡献和支持？
Flink 诞生于欧洲的一个大数据研究项目 StratoSphere。该项目是柏林工业大学的一个研究性项目。早期，Flink 是做 Batch 计算的，但是在 2014 年，StratoSphere 里面的核心成员孵化出 Flink，同年将 Flink 捐赠 Apache，并在后来成为 Apache 的顶级大数据项目，同时 Flink 计算的主流方向被定位为 Streaming, 即用流式计算来做所有大数据的计算，这就是 Flink 技术诞生的背景。
2014 年 Flink 作为主攻流计算的大数据引擎开始在开源大数据行业内崭露头角。区别于 Storm、Spark Streaming 以及其他流式计算引擎的是：它不仅是一个高吞吐、低延迟的计算引擎，同时还提供很多高级的功能。比如它提供了有状态的计算，支持状态管理，支持强一致性的数据语义以及支持 Event Time,WaterMark 对消息乱序的处理。
Flink 核心概念以及基本理念
Flink 最区别于其他流计算引擎的，其实就是状态管理。
什么是状态？例如开发一套流计算的系统或者任务做数据处理，可能经常要对数据进行统计，如 Sum、Count、Min、Max, 这些值是需要存储的。因为要不断更新，这些值或者变量就可以理解为一种状态。如果数据源是在读取 Kafka、RocketMQ，可能要记录读取到什么位置，并记录 Offset，这些 Offset 变量都是要计算的状态。
Flink 提供了内置的状态管理，可以把这些状态存储在 Flink 内部，而不需要把它存储在外部系统。这样做的好处是第一降低了计算引擎对外部系统的依赖以及部署，使运维更加简单；第二，对性能带来了极大的提升：如果通过外部去访问，如 Redis，HBase，它一定是通过网络及 RPC。如果通过 Flink 内部去访问，它只通过自身的进程去访问这些变量。同时 Flink 会定期将这些状态做 Checkpoint 持久化，把 Checkpoint 存储到一个分布式的持久化系统中，比如 HDFS。这样的话，当 Flink 的任务出现任何故障时，它都会从最近的一次 Checkpoint 将整个流的状态进行恢复，然后继续运行它的流处理。对用户没有任何数据上的影响。
Flink 是如何做到在 Checkpoint 恢复过程中没有任何数据的丢失和数据的冗余？来保证精准计算的？
这其中原因是 Flink 利用了一套非常经典的 Chandy-Lamport 算法，它的核心思想是把这个流计算看成一个流式的拓扑，定期从这个拓扑的头部 Source 点开始插入特殊的 Barriers，从上游开始不断的向下游广播这个 Barriers。每一个节点收到所有的 Barriers, 会将 State 做一次 Snapshot，当每个节点都做完 Snapshot 之后，整个拓扑就算完整的做完了一次 Checkpoint。接下来不管出现任何故障，都会从最近的 Checkpoint 进行恢复。
Flink 利用这套经典的算法，保证了强一致性的语义。这也是 Flink 与其他无状态流计算引擎的核心区别。
下面介绍 Flink 是如何解决乱序问题的。比如星球大战的播放顺序，如果按照上映的时间观看，可能会发现故事在跳跃。
在流计算中，与这个例子是非常类似的。所有消息到来的时间，和它真正发生在源头，在线系统 Log 当中的时间是不一致的。在流处理当中，希望是按消息真正发生在源头的顺序进行处理，不希望是真正到达程序里的时间来处理。Flink 提供了 Event Time 和 WaterMark 的一些先进技术来解决乱序的问题。使得用户可以有序的处理这个消息。这是 Flink 一个很重要的特点。
接下来要介绍的是 Flink 启动时的核心理念和核心概念，这是 Flink 发展的第一个阶段；第二个阶段时间是 2015 年和 2017 年，这个阶段也是 Flink 发展以及阿里巴巴介入的时间。故事源于 2015 年年中，我们在搜索事业部的一次调研。当时阿里有自己的批处理技术和流计算技术，有自研的，也有开源的。但是，为了思考下一代大数据引擎的方向以及未来趋势，我们做了很多新技术的调研。
结合大量调研结果，我们最后得出的结论是：解决通用大数据计算需求，批流融合的计算引擎，才是大数据技术的发展方向，并且最终我们选择了 Flink。
但 2015 年的 Flink 还不够成熟，不管是规模还是稳定性尚未经历实践。最后我们决定在阿里内部建立一个 Flink 分支，对 Flink 做大量的修改和完善，让其适应阿里巴巴这种超大规模的业务场景。在这个过程当中，我们团队不仅对 Flink 在性能和稳定性上做出了很多改进和优化，同时在核心架构和功能上也进行了大量创新和改进，并将其贡献给社区，例如：Flink 新的分布式架构，增量 Checkpoint 机制, 基于 Credit-based 的网络流控机制和 Streaming SQL 等。
阿里巴巴对 Flink 社区的贡献
我们举两个设计案例，第一个是阿里巴巴重构了 Flink 的分布式架构，将 Flink 的 Job 调度和资源管理做了一个清晰的分层和解耦。这样做的首要好处是 Flink 可以原生的跑在各种不同的开源资源管理器上。经过这套分布式架构的改进，Flink 可以原生地跑在 Hadoop Yarn 和 Kubernetes 这两个最常见的资源管理系统之上。同时将 Flink 的任务调度从集中式调度改为了分布式调度，这样 Flink 就可以支持更大规模的集群，以及得到更好的资源隔离。
另一个是实现了增量的 Checkpoint 机制，因为 Flink 提供了有状态的计算和定期的 Checkpoint 机制，如果内部的数据越来越多，不停地做 Checkpoint, Checkpoint 会越来越大，最后可能导致做不出来。提供了增量的 Checkpoint 后，Flink 会自动地发现哪些数据是增量变化，哪些数据是被修改了。同时只将这些修改的数据进行持久化。这样 Checkpoint 不会随着时间的运行而越来越难做，整个系统的性能会非常地平稳，这也是我们贡献给社区的一个很重大的特性。
经过 2015 年到 2017 年对 Flink Streaming 的能力完善，Flink 社区也逐渐成熟起来。Flink 也成为在 Streaming 领域最主流的计算引擎。因为 Flink 最早期想做一个流批统一的大数据引擎，2018 年已经启动这项工作，为了实现这个目标，阿里巴巴提出了新的统一 API 架构，统一 SQL 解决方案，同时流计算的各种功能得到完善后，我们认为批计算也需要各种各样的完善。无论在任务调度层，还是在数据 Shuffle 层，在容错性，易用性上，都需要完善很多工作。
篇幅原因，下面主要和大家分享两点：
- 统一 API Stack
- 统一 SQL 方案
先来看下目前 Flink API Stack 的一个现状，调研过 Flink 或者使用过 Flink 的开发者应该知道。Flink 有 2 套基础的 API，一套是 DataStream，一套是 DataSet。DataStream API 是针对流式处理的用户提供，DataSet API 是针对批处理用户提供，但是这两套 API 的执行路径是完全不一样的，甚至需要生成不同的 Task 去执行。所以这跟得到统一的 API 是有冲突的，而且这个也是不完善的，不是最终的解法。在 Runtime 之上首先是要有一个批流统一融合的基础 API 层，我们希望可以统一 API 层。
因此，我们在新架构中将采用一个 DAG（有限无环图）API，作为一个批流统一的 API 层。对于这个有限无环图，批计算和流计算不需要泾渭分明的表达出来。只需要让开发者在不同的节点，不同的边上定义不同的属性，来规划数据是流属性还是批属性。整个拓扑是可以融合批流统一的语义表达，整个计算无需区分是流计算还是批计算，只需要表达自己的需求。有了这套 API 后，Flink 的 API Stack 将得到统一。
除了统一的基础 API 层和统一的 API Stack 外，同样在上层统一 SQL 的解决方案。流和批的 SQL，可以认为流计算有数据源，批计算也有数据源，我们可以将这两种源都模拟成数据表。可以认为流数据的数据源是一张不断更新的数据表，对于批处理的数据源可以认为是一张相对静止的表，没有更新的数据表。整个数据处理可以当做 SQL 的一个 Query，最终产生的结果也可以模拟成一个结果表。
对于流计算而言，它的结果表是一张不断更新的结果表。对于批处理而言，它的结果表是相当于一次更新完成的结果表。从整个 SQL 语义上表达，流和批是可以统一的。此外，不管是流式 SQL，还是批处理 SQL，都可以用同一个 Query 来表达复用。这样以来流批都可以用同一个 Query 优化或者解析。甚至很多流和批的算子都是可以复用的。
Flink 的未来方向
首先，阿里巴巴还是要立足于 Flink 的本质，去做一个全能的统一大数据计算引擎。将它在生态和场景上进行落地。目前 Flink 已经是一个主流的流计算引擎，很多互联网公司已经达成了共识：Flink 是大数据的未来，是最好的流计算引擎。下一步很重要的工作是让 Flink 在批计算上有所突破。在更多的场景下落地，成为一种主流的批计算引擎。然后进一步在流和批之间进行无缝的切换，流和批的界限越来越模糊。用 Flink, 在一个计算中，既可以有流计算，又可以有批计算。
第二个方向就是 Flink 的生态上有更多语言的支持，不仅仅是 Java，Scala 语言，甚至是机器学习下用的 Python，Go 语言。未来我们希望能用更多丰富的语言来开发 Flink 计算的任务，来描述计算逻辑，并和更多的生态进行对接。
最后不得不说 AI，因为现在很多大数据计算的需求和数据量都是在支持很火爆的 AI 场景，所以在 Flink 流批生态完善的基础上，将继续往上走，完善上层 Flink 的 Machine Learning 算法库，同时 Flink 往上层也会向成熟的机器学习，深度学习去集成。比如可以做 Tensorflow On Flink, 让大数据的 ETL 数据处理和机器学习的 Feature 计算和特征计算，训练的计算等进行集成，让开发者能够同时享受到多种生态给大家带来的好处。