Flink：特性、概念、组件栈、架构及原理分析

简单之美 | Apache Flink：特性、概念、组件栈、架构及原理分析

http://shiyanjun.cn/archives/1508.html

Apache Flink是一个面向分布式数据流处理和批量数据处理的开源计算平台，它能够基于同一个Flink运行时（Flink Runtime），提供支持流处理和批处理两种类型应用的功能。现有的开源计算方案，会把流处理和批处理作为两种不同的应用类型，因为他们它们所提供的SLA是完全不相同的：流处理一般需要支持低延迟、Exactly-once保证，而批处理需要支持高吞吐、高效处理，所以在实现的时候通常是分别给出两套实现方法，或者通过一个独立的开源框架来实现其中每一种处理方案。例如，实现批处理的开源方案有MapReduce、Tez、Crunch、Spark，实现流处理的开源方案有Samza、Storm。Flink在实现流处理和批处理时，与传统的一些方案完全不同，它从另一个视角看待流处理和批处理，将二者统一起来：Flink是完全支持流处理，也就是说作为流处理看待时输入数据流是无界的；批处理被作为一种特殊的流处理，只是它的输入数据流被定义为有界的。基于同一个Flink运行时（Flink Runtime），分别提供了流处理和批处理API，而这两种API也是实现上层面向流处理、批处理类型应用框架的基础。

基本特性

关于Flink所支持的特性，我这里只是通过分类的方式简单做一下梳理，涉及到具体的一些概念及其原理会在后面的部分做详细说明。

流处理特性

支持高吞吐、低延迟、高性能的流处理

支持带有事件时间的窗口（Window） *** 作

支持有状态计算的Exactly-once语义

支持高度灵活的窗口（Window） *** 作，支持基于time、count、session，以及data-driven的窗口 *** 作

支持具有Backpressure功能的持续流模型

支持基于轻量级分布式快照（Snapshot）实现的容错

一个运行时同时支持Batch on Streaming处理和Streaming处理

Flink在JVM内部实现了自己的内存管理

支持迭代计算

支持程序自动优化：避免特定情况下Shuffle、排序等昂贵 *** 作，中间结果有必要进行缓存

API支持

对Streaming数据类应用，提供DataStream API

对批处理类应用，提供DataSet API（支持Java/Scala）

Libraries支持

支持机器学习（FlinkML）

支持图分析（Gelly）

支持关系数据处理（Table）

支持复杂事件处理（CEP）

整合支持

支持Flink on YARN

支持HDFS

支持来自Kafka的输入数据

支持Apache HBase

支持Hadoop程序

支持Tachyon

支持ElasticSearch

支持RabbitMQ

支持Apache Storm

支持S3

支持XtreemFS

基本概念

Stream &Transformation &Operator

用户实现的Flink程序是由Stream和Transformation这两个基本构建块组成，其中Stream是一个中间结果数据，而Transformation是一个 *** 作，它对一个或多个输入Stream进行计算处理，输出一个或多个结果Stream。当一个Flink程序被执行的时候，它会被映射为Streaming Dataflow。一个Streaming Dataflow是由一组Stream和Transformation Operator组成，它类似于一个DAG图，在启动的时候从一个或多个Source Operator开始，结束于一个或多个Sink Operator。下面是一个由Flink程序映射为Streaming Dataflow的示意图，如下所示：

比如从Source[1]到map()[1]，它保持了Source的分区特性（Partitioning）和分区内元素处理的有序性，也就是说map()[1]的Subtask看到数据流中记录的顺序，与Source[1]中看到的记录顺序是一致的。

Redistribution模式

这种模式改变了输入数据流的分区，比如从map()[1]、map()[2]到keyBy()/window()/apply()[1]、keyBy()/window()/apply()[2]，上游的Subtask向下游的多个不同的Subtask发送数据，改变了数据流的分区，这与实际应用所选择的Operator有关系。另外，Source Operator对应2个Subtask，所以并行度为2，而Sink Operator的Subtask只有1个，故而并行度为1。

Task &Operator Chain

在Flink分布式执行环境中，会将多个Operator Subtask串起来组成一个Operator Chain，实际上就是一个执行链，每个执行链会在TaskManager上一个独立的线程中执行，如下图所示：

在Flink集群启动的时候，TaskManager会向JobManager注册，如果注册成功，则JobManager会向TaskManager回复消息AcknowledgeRegistration。

SubmitJob

Flink程序内部通过Client向JobManager提交Flink Job，其中在消息SubmitJob中以JobGraph形式描述了Job的基本信息。

CancelJob

请求取消一个Flink Job的执行，CancelJob消息中包含了Job的ID，如果成功则返回消息CancellationSuccess，失败则返回消息CancellationFailure。

UpdateTaskExecutionState

TaskManager会向JobManager请求更新ExecutionGraph中的ExecutionVertex的状态信息，更新成功则返回true。

RequestNextInputSplit

运行在TaskManager上面的Task，请求获取下一个要处理的输入Split，成功则返回NextInputSplit。

JobStatusChanged

ExecutionGraph向JobManager发送该消息，用来表示Flink Job的状态发生的变化，例如：RUNNING、CANCELING、FINISHED等。

TaskManager

TaskManager也是一个Actor，它是实际负责执行计算的Worker，在其上执行Flink Job的一组Task。每个TaskManager负责管理其所在节点上的资源信息，如内存、磁盘、网络，在启动的时候将资源的状态向JobManager汇报。TaskManager端可以分成两个阶段：

注册阶段

TaskManager会向JobManager注册，发送RegisterTaskManager消息，等待JobManager返回AcknowledgeRegistration，然后TaskManager就可以进行初始化过程。

可 *** 作阶段

该阶段TaskManager可以接收并处理与Task有关的消息，如SubmitTask、CancelTask、FailTask。如果TaskManager无法连接到JobManager，这是TaskManager就失去了与JobManager的联系，会自动进入“注册阶段”，只有完成注册才能继续处理Task相关的消息。

Client

当用户提交一个Flink程序时，会首先创建一个Client，该Client首先会对用户提交的Flink程序进行预处理，并提交到Flink集群中处理，所以Client需要从用户提交的Flink程序配置中获取JobManager的地址，并建立到JobManager的连接，将Flink Job提交给JobManager。Client会将用户提交的Flink程序组装一个JobGraph， 并且是以JobGraph的形式提交的。一个JobGraph是一个Flink Dataflow，它由多个JobVertex组成的DAG。其中，一个JobGraph包含了一个Flink程序的如下信息：JobID、Job名称、配置信息、一组JobVertex等。

组件栈

Flink是一个分层架构的系统，每一层所包含的组件都提供了特定的抽象，用来服务于上层组件。Flink分层的组件栈如下图所示：

了解YARN的话，对上图的原理非常熟悉，实际Flink也实现了满足在YARN集群上运行的各个组件：Flink YARN Client负责与YARN RM通信协商资源请求，Flink JobManager和Flink TaskManager分别申请到Container去运行各自的进程。通过上图可以看到，YARN AM与Flink JobManager在同一个Container中，这样AM可以知道Flink JobManager的地址，从而AM可以申请Container去启动Flink TaskManager。待Flink成功运行在YARN集群上，Flink YARN Client就可以提交Flink Job到Flink JobManager，并进行后续的映射、调度和计算处理。

Runtime层

Runtime层提供了支持Flink计算的全部核心实现，比如：支持分布式Stream处理、JobGraph到ExecutionGraph的映射、调度等等，为上层API层提供基础服务。

API层

API层主要实现了面向无界Stream的流处理和面向Batch的批处理API，其中面向流处理对应DataStream API，面向批处理对应DataSet API。

Libraries层

该层也可以称为Flink应用框架层，根据API层的划分，在API层之上构建的满足特定应用的实现计算框架，也分别对应于面向流处理和面向批处理两类。面向流处理支持：CEP（复杂事件处理）、基于SQL-like的 *** 作（基于Table的关系 *** 作）；面向批处理支持：FlinkML（机器学习库）、Gelly（图处理）。

内部原理

容错机制

Flink基于Checkpoint机制实现容错，它的原理是不断地生成分布式Streaming数据流Snapshot。在流处理失败时，通过这些Snapshot可以恢复数据流处理。理解Flink的容错机制，首先需要了解一下Barrier这个概念：Stream Barrier是Flink分布式Snapshotting中的核心元素，它会作为数据流的记录被同等看待，被插入到数据流中，将数据流中记录的进行分组，并沿着数据流的方向向前推进。每个Barrier会携带一个Snapshot ID，属于该Snapshot的记录会被推向该Barrier的前方。因为Barrier非常轻量，所以并不会中断数据流。带有Barrier的数据流，如下图所示：

接收到Barrier n的Stream被临时搁置，来自这些Stream的记录不会被处理，而是被放在一个Buffer中

一旦最后一个Stream接收到Barrier n，Operator会emit所有暂存在Buffer中的记录，然后向Checkpoint Coordinator发送Snapshot n

继续处理来自多个Stream的记录

基于Stream Aligning *** 作能够实现Exactly Once语义，但是也会给流处理应用带来延迟，因为为了排列对齐Barrier，会暂时缓存一部分Stream的记录到Buffer中，尤其是在数据流并行度很高的场景下可能更加明显，通常以最迟对齐Barrier的一个Stream为处理Buffer中缓存记录的时刻点。在Flink中，提供了一个开关，选择是否使用Stream Aligning，如果关掉则Exactly Once会变成At least once。

调度机制

在JobManager端，会接收到Client提交的JobGraph形式的Flink Job，JobManager会将一个JobGraph转换映射为一个ExecutionGraph，如下图所示：

迭代机制

机器学习和图计算应用，都会使用到迭代计算，Flink通过在迭代Operator中定义Step函数来实现迭代算法，这种迭代算法包括Iterate和Delta Iterate两种类型，在实现上它们反复地在当前迭代状态上调用Step函数，直到满足给定的条件才会停止迭代。下面，对Iterate和Delta Iterate两种类型的迭代算法原理进行说明：

Iterate

Iterate Operator是一种简单的迭代形式：每一轮迭代，Step函数的输入或者是输入的整个数据集，或者是上一轮迭代的结果，通过该轮迭代计算出下一轮计算所需要的输入（也称为Next Partial Solution），满足迭代的终止条件后，会输出最终迭代结果，具体执行流程如下图所示：

Delta Iterate Operator实现了增量迭代，它的实现原理如下图所示：

另外，Flink还提供了3个参数来配置Backpressure监控行为：

参数名称

默认值

说明

jobmanager.web.backpressure.refresh-interval

60000

默认1分钟，表示采样统计结果刷新时间间隔

jobmanager.web.backpressure.num-samples

100

评估Backpressure状态，所使用的堆栈跟踪调用次数

jobmanager.web.backpressure.delay-between-samples

50

默认50毫秒，表示对一个Job的每个Task依次调用的时间间隔

通过上面个定义的Backpressure状态，以及调整相应的参数，可以确定当前运行的Job的状态是否正常，并且保证不影响JobManager提供服务。

参考链接

http://flink.apache.org/

http://flink.apache.org/features.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/general_arch.html

http://data-artisans.com/high-throughput-low-latency-and-exactly-once-stream-processing-with-apache-flink/

https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/stream_checkpointing.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/job_scheduling.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/streaming/event_time.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/yarn_setup.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/jobmanager_high_availability.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/streaming/libs/cep.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/batch/libs/gelly.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/batch/libs/ml/index.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/batch/libs/table.html

https://ci.apache.org/projects/flink/flink-docs-release-1.0/apis/batch/connectors.html

http://geek.csdn.net/news/detail/56272

http://samza.apache.org/

在 flink 任务运行的过程中，用户通常想知道任务运行的一些基本指标，比如吞吐量、内存和 cpu 使用情况、checkpoint 稳定性等等。而通过 flink metrics 这些指标都可以轻而易举地获取到，避免任务的运行处于黑盒状态，通过分析这些指标，可以更好的调整任务的资源、定位遇到的问题、对任务进行监控。接下来本文将介绍 flink metrics 的一些基本概念与原理以及实践。

Flink 对于指标监测有一套自己的实现，同时 flink 自身系统有一些固定的 metric 数据，　包括系统的一些指标，CPU,内存, IO 　或者各个 task 运行的一些指标。指标的统计方式有四种，这些指标都实现了 Metric 这个接口，而 Metric 这个接口只是一个标识，本身并没有定义如何方法接口，部分子类的继承关系如下所示。

从图中可以看出，Metric 这个接口有四个直接子类，分别是：

下面以 Counter 为例，说明 Metric 的具体用法，Counters 通常用来计数，可以通过 inc 或 dec 方法来对计数值进行增加或减少。

获取 Metrics 有三种方法，首先可以在 WebUI 上看到；其次可以通过 RESTful API 获取，RESTful API 对程序比较友好，比如写自动化脚本或程序，自动化运维和测试，通过 RESTful API 解析返回的 Json 格式对程序比较友好；最后，还可以通过 Metric Reporter 获取，监控主要使用 Metric Reporter 功能。

flink 提供了很多外部监控系统的支持：JMX(java 自带的技术，不严格属于第三方)、Graphite、InfluxDB、Prometheus、StatsD、Datadog、Slf4j（直接打 log 里）等，也可以通过实现 org.apache.flink.metrics.reporter.MetricReporter 接口来编写自己的 Reporter。如果想要定期发送报告，可以实现 Scheduled 接口。

Metric Reporter 是如何配置的？首先 Metrics Reporters 的名字用逗号分隔，然后通过 metrics.reporter.jmx.class 的 classname 反射找 reporter，还需要拿到 metrics.reporter.jmx.port 的配置，比如向第三方系统通过网络发送的比较多，但要知道往哪里发，ip 地址、port 信息是比较常见的。

开发者可以实现自己的 reporter，将 metrics 数据导出到不同的系统。

MetricReporter 是用来向外暴露 Metric 的监测结果的接口。由于 MetricReporter 的子类在实例化时，都是通过反射机制，所以对于其实现子类，需要有一个公共、无参的构造函数，这个接口的定义如下：

关注 gzh “HEY DATA” 后台回复关键字 MetricReporter 可获得自定义 MetricReporter 实现例子文件。

---