Apache Flink® - 数据流上的有状态计算

Apache Flink 是为分布式、高性能、随 时可用以及准确的流处理应用程序打造的开源流处理框架” 。 Apache Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。 Flink 被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。
[successbox title="Flink 流处理特性"]
1.支持高吞吐、 低延迟、 高性能的流处理
2.支持带有事件时间的窗口（Window） 操作
3.支持有状态计算的 Exactly-once 语义
4.支持高度灵活的窗口（Window） 操作， 支持基于 time、 count、 session，以及 data-driven 的窗口操作
5.支持具有 Backpressure 功能的持续流模型
6.支持基于轻量级分布式快照（Snapshot） 实现的容错
7.一个运行时同时支持 Batch on Streaming 处理和 Streaming 处理
8.Flink 在 JVM 内部实现了自己的内存管理
9.支持迭代计算
10.支持程序自动优化： 避免特定情况下 Shuffle、 排序等昂贵操作， 中间结果有必要进行缓存
[/successbox]
[successbox title="Flink的四大基石"]

checkpoint,State,Time,Window
[/successbox]
[successbox title="批处理与流处理"]
批处理的特点是有界、 持久、 大量， 批处理非常适合需要访问全套记录才能完成的计算工作， 一般用于离线统计 。
流处理的特点是无界、 实时， 流处理方式无需针对整个数据集执行操作， 而是对通过系统传输的每个数据项执行操作， 一般用于实时统计。
在 Spark 生态体系中， 对于批处理和流处理采用了不同的技术框架， 批处理由SparkSQL 实现， 流处理由 Spark Streaming 实现， 这也是大部分框架采用的策略， 使 用独立的处理器实现批处理和流处理， 而 Flink 可以同时实现批处理和流处理。
Flink 将批处理（ 即处理有 限的静态数据）视作一种特殊的流处理。

Flink 分别提供了面向流式 处理的接口（DataStream API） 和面向批处理的接口（DataSet API） 。
因此，Flink 既 可以完成流处理，也可以完成批处理。Flink 支持的拓展库涉及机器学习（FlinkML）、复杂事件处理（CEP）、以及图计算（Gelly），还有分别针对流处理和批处理的 Table API。
[/successbox]

[successbox title="Flink的角色"]

JobManager 处理器： 也称之为 Master， 用于协调分布式执行， 它们用来调度 task， 协调检查点， 协调失败 时恢复等。 Flink 运行时至少存在一个 master 处理器， 如果配置高可用模式则会存在多 个 master 处理器， 它们其中有一个是 leader， 而其他的都是 standby。 TaskManager 处理器： 也称之为 Worker， 用于执行一个 dataflow 的 task(或者特殊的 subtask)、 数据缓冲和 datastream 的交换， Flink 运行时至少会存在一个 worker 处理器。
[/successbox]
[successbox title="无界数据流与有界数据流"]
任何类型的数据都可以形成一种事件流。信用卡交易、传感器测量、机器日志、网站或移动应用程序上的用户交互记录，所有这些数据都形成一种流。
数据可以被作为 无界 或者 有界 流来处理。
无界流 有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。无界数据流不会在生成时终止并提供数据，必须连续处理无界流， 也就是说必须在获取后立即处理 event。对于无界数据流我 们无法等待所有数据都到达， 因为输入是无界的， 并且在任何时间点都不会完成。 处理 无界数据通常要求以特定顺序（ 例如事件发生的顺序） 获取 event，以便能够推断结果 完整性。

有界流 有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。因为可以始终对有界数据集进行排序，有界流处理通常被称为批处理，可以在执行任何计算之前通过获取所有数据来处理有界流， 处理有界流不需要有序获取。

流处理一般需要支持低延迟、 Exactly-once 保证， 而批处 理需要支持高吞吐、 高效处理
Flink 是完全支持流处理， 也就是说作为流处理看 待时输入数据流是无界的； 批处理被作为一种特殊的流处理， 只是它的输入数据流被定义 为有界的。 基于同一个 Flink 运行时(Flink Runtime)， 分别提供了流处理和批处理 API， 而这两种 API 也是实现上层面向流处理、 批处理类型应用框架的基础。
[/successbox]
[successbox title="Flink数据流编程模型"]

• 最底层的抽象仅提供状态流（stateful streaming），它通过处理函数嵌入到DataStream API中。
• 实践中用Core API比较多，这些流式的API提供了通用的构建入口用于数据处理，像各种用户自定义的transformation、join、aggregation、window、state等。
• Table API是以表为中心的声明式DSL（领域特定语言），当这些Table表示的是stream时，Table是动态变化的。Table API遵循扩展的关系模型，提供了包括select、project、join、group-by、aggregate等操作。可 以 在 表 与 DataStream/DataSet 之 间 无 缝 切 换 ，以 允 许 程 序 将 Table API 与 DataStream 以及 DataSet 混合使用。
• Flink提供的最高层级的API是SQL，它在语义和表达能力上与Table API是类似的。

[/successbox]
[successbox title="Libraries 支持"]
支持机器学习（ FlinkML）
支持图分析（ Gelly）
支持关系数据处理（ Table）
支持复杂事件处理（ CEP）
[/successbox]

[successbox title="任务提交流程"]

Flink任务提交后，Client向HDFS上传Flink的Jar包和配置，之后向Yarn
ResourceManager提交任务，ResourceManager分配Container资源并通知对
应的NodeManager启动ApplicationMaster，ApplicationMaster启动后加载Flink的
Jar包和配置构建环境，然后启动JobManager，之后ApplicationMaster向
ResourceManager申请资源启动TaskManager，ResourceManager分配Container
资源后，由ApplicationMaster通知资源所在节点的NodeManager启动
TaskManager，NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager，
TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务。
-------------------------
1.Client 向 HDFS 上传 Flink 的 Jar 包和配置
2.Client向 YarnResourceManager 提 交 任 务
3.ResourceManager 分 配 Container 资 源 并 通 知 对应 的 NodeManager 启 动 ApplicationMaster
4.ApplicationMaster 启动后加载 Flink 的Jar 包 和 配 置 构 建 环 境 ， 然 后 启 动 JobManager
5.ApplicationMaster 向ResourceManager 申 请 资 源 启 动 TaskManager
6.ResourceManager 分 配 Container 源 后 ， 由 ApplicationMaster 通 知 资 源 所 在 节 点 的 NodeManager 启动TaskManager
7.NodeManager 加载 Flink 的 Jar 包和配置构建环境并启动 TaskManager
8.TaskManager 启动后向 JobManager 发送心跳包， 并等待 JobManager 向其分配任务
[/successbox]
[successbox title="Worker 与 Slots"]
每一个worker(TaskManager)是一个JVM进程，它可能会在独立的线程上执行一个或
多个subtask。为了控制一个worker能接收多少个task，worker通过taskslot来进
行控制（一个worker至少有一个taskslot）。
每个taskslot表示TaskManager拥有资源的一个固定大小的子集。假如一个
TaskManager有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化
意味着一个subtask将不需要跟来自其他job的subtask竞争被管理的内存，取而代之
的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU的隔离，slot目
前仅仅用来隔离task的受管理的内存。
通过调整taskslot的数量，允许用户定义subtask之间如何互相隔离。如果一个
TaskManager一个slot，那将意味着每个taskgroup运行在独立的JVM中（该JVM可
能是通过一个特定的容器启动的），而一个TaskManager多个slot意味着更多的subtask
可以共享同一个JVM。而在同一个JVM进程中的task将共享TCP连接（基于多路复用）
和心跳消息。它们也可能共享数据集和数据结构，因此这减少了每个task的负载。

TaskSlot是静态的概念，是指TaskManager具有的并发执行能力，可以通过参数
taskmanager.numberOfTaskSlots进行配置，而并行度parallelism是动态概念，即
TaskManager运行程序时实际使用的并发能力，可以通过参数parallelism.default进行
配置。也就是说，假设一共有3个TaskManager，每一个TaskManager中的分配3个
TaskSlot，也就是每个TaskManager可以接收3个task，一共9个
TaskSlot，如果我们设置parallelism.default=1，即运行程序默认的并行度为1，
9个TaskSlot只用了1个，有8个空闲，因此，设置合适的并行度才能提高效率。
------------------------------
每个 task slot 表示 TaskManager 拥有资源的一个固定大小的子集。 假如一个
TaskManager 有三个 slot， 那么它会将其管理的内存分成三份给各个 slot。 资源 slot 化
意味着一个 subtask 将不需要跟来自其他 job 的 subtask 竞争被管理的内存， 取而代之
的是它将拥有一定数量的内存 储备。 需要注意的是， 这里不会涉及到 CPU 的隔离， slot 目前仅仅用来隔离 task 的受管理的内存 。
[/successbox]
[successbox title="程序与数据流"]
Flink程序的基础构建模块是流（streams）与转换（transformations）（需要注意
的是，Flink的DataSetAPI所使用的DataSets其内部也是stream）。一个stream可
以看成一个中间结果，而一个transformations是以一个或多个stream作为输入的某种
operation，该operation利用这些stream进行计算从而产生一个或多个resultstream。
在运行时，Flink上运行的程序会被映射成streamingdataflows，它包含了streams
和transformationsoperators。每一个dataflow以一个或多个sources开始以一个或
多个sinks结束。dataflow类似于任意的有向无环图（DAG），当然特定形式的环可以通过
iteration构建。在大部分情况下，程序中的transformations跟dataflow中的
operator是一一对应的关系，但有时候，一个transformation可能对应多个operator。

[/successbox]
[successbox title="并行数据流"]
Flink程序的执行具有并行、分布式的特性。在执行过程中，一个stream包含一个或
多个streampartition，而每一个operator包含一个或多个operatorsubtask，这些
operatorsubtasks在不同的线程、不同的物理机或不同的容器中彼此互不依赖得执行。
一个特定operator的subtask的个数被称之为其parallelism(并行度)。一个
stream的并行度总是等同于其producingoperator的并行度。一个程序中，不同的
operator可能具有不同的并行度。

Stream在operator之间传输数据的形式可以是one-to-one(forwarding)的模式也
可以是redistributing的模式，具体是哪一种形式，取决于operator的种类。
One-to-one：stream(比如在source和mapoperator之间)维护着分区以及元素的顺
序。那意味着mapoperator的subtask看到的元素的个数以及顺序跟sourceoperator
的subtask生产的元素的个数、顺序相同，map、fliter、flatMap等算子都是one-to-one
的对应关系。
Redistributing：stream(map()跟keyBy/window之间或者keyBy/window跟sink之
间)的分区会发生改变。每一个operatorsubtask依据所选择的transformation发送数
据到不同的目标subtask。例如，keyBy()基于hashCode重分区、broadcast和rebalance
会随机重新分区，这些算子都会引起redistribute过程，而redistribute过程就类似
于Spark中的shuffle过程。
[/successbox]
[successbox title="task 与 operator chains"]
出于分布式执行的目的，Flink将operator的subtask链接在一起形成task，每个
task在一个线程中执行。将operators链接成task是非常有效的优化：它能减少线程之
间的切换和基于缓存区的数据交换，在减少时延的同时提升吞吐量。链接的行为可以在编程
API中进行指定。
下面这幅图，展示了5个subtask以5个并行的线程来执行：

[/successbox]