玩转Flink流批一体分布式实时处理引擎

Apache Flink是为分布式、高性能的流处理应用程序打造的开源流处理框架。Flink不仅能提供同时支持高吞吐和exactly-once语义的实时计算，还能提供批量数据处理。相较于市面上的其他数据处理引擎，它采用的是基于流计算来模拟批处理。

一、Flink原理及架构

Flink简介

Apache Flink是为分布式、高性能的流处理应用程序打造的开源流处理框架。Flink不仅能提供同时支持高吞吐和exactly-once语义的实时计算，还能提供批量数据处理。主要由Java代码实现，支持实时流处理和批处理，批数据只是流数据的一个极限案例。支持了迭代计算，内存管理和程序优化。

相较于市面上的其他数据处理引擎，Flink和Spark都可以同时支持流处理和批处理。但是,Spark的技术理念是基于批处理来模拟流的计算;而Flink则完全相反，它采用的是基于流计算来模拟批处理。

Flink关键机制

四个机制:状态、时间、检查点、窗口

Flink中有四种最重要的关键机制，这些关键机制在后面我们也会来进行详细的介绍，这里我们主要介绍它的基本概念以及主要用途。首先Flink中最重要的一个机制是状态机制(State)，Flink是一种有状态的流计算引擎。状态的作用主要是我们Flink是一种流计算，它需要存储节点的中间计算结果。另外状态的保存还有利于Flink进行容错恢复。状态有密切关系的是Flink的Checkpoint，也就是检查点的机制，Checkpoint能够去把Flink的状态进行存储，相当于是做一次快照，方便Flink进行容错恢复。另外因为Flink它是一种流计算引擎，它的数据是不间断产生的，是没有界限的，因此我们需要有一种机制能够对数据进行切分，我们会采用的时间(Time)作为切分点，另外Flink进行容错性的恢复，它也需要知道从哪个时间点来进行恢复。所以说时间也是Flink中一种很重要的机制。最后是窗口window，在Flink中需要使用的窗口对数据进行切分，也方便对数据进行聚合计算。

Flink核心理念

Flink与其他流计算引擎的最大区别，就是状态管理。

Flink提供了内置的状态管理，可以把工作时状态存储在Flink内部，而不需要把它存储在外部系统。这样做的好处:

• 降低了计算引擎对外部系统的依赖，使得部署、运维更加简单;
• 对性能带来了极大的提升。

Flink Runtime整体架构

Flink运行时架构从下至上可以分为了三层，在最下层是Flink的一些配置方式，Flink可以采用单机的方式安装，也可以采用的集群的方式安装，另外也可以采用云的方式部署。在大多数情况下，Flink都是采用的集群的方式进行配置和安装的。其中呢它支持了两种集群模式，一种是Standalon，这种方式是采用了Flink自身提供的资源调度管理器。另外一种方式是基于YARN的方式进行了配置安装。

YARN提供了专用的资源管理器。在中间层次是Flink的计算引擎，这个计算引擎它同时能够支持流处理和批处理，可以接收了上层的api提交给它做作业 。Runtime这个引擎上面可以分为了两个模块，一个模块是DataStream api，一个是DataSet api。Flink向dataset和datastream，也就是批数据集以及流数据集是分开处理的，但是都是公用下面的计算引擎。基于两种类型的api，Flink又提供了更多的上层的抽象的api，API越抽象，它的表达能力越弱，但是它对数据的处理能力、抽象性也越强。在针对于上层Table api和SQL，它是主要是针对关系运算的，那针对关系数据的查询，Flink提供了统一的接口，基于流数据api，同时提供了复杂事件处理api。复杂事件指的就是说对不能够用时间去表示事件的开始、次序以及结束这样的事件进行处理的api接口。另外针对于数据及api，它提供了机器学习api以及图计算的api。

Flink核心概念- DataStream

DataStream: Flink用类DataStream来表示程序中的流式数据。用户可以认为它们是含有重复数据的不可修改的集合(collection)，DataStream中元素的数量是无限的。

从图中我们可以发现，对DataStream可以使用一些算子，例如KeyBy这样的算子，对它进行处理转换之后，它会转换成另外一种数据流，也称为keyedstream。那么基于keyedstream，我们进一步可以使用窗口算子，这主要是Flink程序设计中对数据流的一些处理方式。

Flink核心概念- DataSet

DataSet : Flink系统可对数据集进行转换（例如，过滤，映射，联接，分组)，数据集可从读取文件或从本地集合创建。结果通过接收器( Sink)返回，接收器可以将数据写入(分布式）文件或标准输出（例如命令行终端)

Flink程序

Flink程序由Source、Transformation和Sink三部分组成，其中Source主要负责数据的读取，支持HDFS、kafka和文本等;Transformation主要负责对数据的转换操作; Sink负责最终数据的输出，支持HDFS、kafka和文本输出等。在各部分之间流转的数据称为流( stream ) 。

Flink数据源

批处理：

Files：HDFS,Local file system，MapR file system；Text，CSV，Avro，Hadoop input formats

JDBC、HBase和 Collections

流处理：

Files、Socket streams、Kafka、RabbitMQ、Flume、Collections、 Implement your own和SourceFunction.collecto

Flink程序运行图

Flink是一种master-Slave架构，它在启动的时候就会产生了JobManger以及TaskManager。事实上在Flink程序中还包含两个组件，这两个组件一个叫resource manager，主要负责了资源的调度与管理，另外一个称为Dispatcher。主要是用来进行client，要把JobManager进行分发公布。我们来看一看具体的运行流程。

首先是用户提交Flink程序，这个Flink程序就会转换成逻辑数据流图。客户端接收到逻辑数据流图之后，然后连同jar包以及一些依赖包就会提交给了JobManger，JobManger接收到逻辑数据流图之后会转成物理数据流图，这个物理数据流图是真实的可执行的，能够具体的将任务放置在TaskManager上，在TaskManager中会将它所拥有的资源划分成一个一个的TaskSlot。每个TaskSlot实际上就相当于是jvm，它的一个具体的线程。每个TaskSlot占用了TaskManager的一部分资源，这里的资源主要是以内存进行划分的，TaskSlot不对cpu的资源进行划分，因此没有对cpu的资源进行隔离。

Flink作业运行流程(一)

用户首先提交Flink程序到JobClient，经过JobClient的处理、解析、优化提交到JobManager，最后由TaskManager运行task。

在Flink中它通过了JobClient提交了任务，做过JobClient提交的任务进一步的进行优化、解析以及处理，提交给了JobManager。JobManager会将jobClient提交了逻辑数据流图转换成物理数据流图，然后将这些任务分配给taskmanager。taskmanager接受到任务之后就相应地进行处理，并且汇报了task的状态给JobManager，JobManager最后就把结果反馈给jobClient。

JobClient是Flink程序和JobManager交互的桥梁。主要负责接收程序、解析程序的执行计划、优化程序的执行计划，然后提交执行计划到JobManager。在Flink中主要有三类Operator。

Source Operator:数据源操作，比如文件、socket、Kafka等。

Transformation Operator:数据转换操作，比如map，flatMap，reduce等算子。

Sink Operator:数据存储操作。比如数据存储到HDFS、Mysql、Kafka等等。

一个完整的Flink程序---java

Flink的数据处理

Apache Flink它同时支持批处理和流处理，也能用来做一些基于事件的应用。

首先Flink是一个纯流式的计算引擎，它的基本数据模型是数据流。流可以是无边界的无限流，即一般意义上的流处理。也可以是有边界的有限流，就是批处理。因此Flink用一套架构同时支持了流处理和批处理。

其次，Flink的一个优势是支持有状态的计算。如果处理一个事件(或一条数据）的结果只跟事6件本身的内容有关，称为无状态处理;反之结果还和之前处理过的事件有关，称为有状态处理。

有界流与无界流

无界流:有定义流的开始，但没有定义流的结束。数据源会无休止地产生数据。无界流的数据必须持续处理,即数据被读取后需要立刻处理。不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

有界流:有定义流的开始，也有定义流的结束。有界流可以在读取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理。

批处理示例

批处理是流处理的一种非常特殊的情况。在流处理中，我们为数据定义滑动窗口或滚动窗口，并且在每次窗口滑动或滚动时生成结果。批处理则不同，我们定义一个全局窗口，所有的记录都属于同一个窗口。举例来说，以下代码表示一个简单的Flink程序，它负责每小时对某网站的访问者计数，并按照地区分组。

如果知道输入数据是有限的，则可以通过以下代码实现批处理。

如果输入数据是有限的，那么下面代码与上面代码的运行结果相同。

Flink批处理模型

Flink通过一个底层引擎同时支持流处理和批处理。

在流处理引擎之上，Flink 有以下机制：

• 检查点机制和状态机制：用于实现容错、有状态的处理；
• 水印机制：用于实现事件时钟；
• 窗口和触发器：用于限制计算范围，并定义呈现结果的时间。

在同一个流处理引擎之上，Flink 还存在另一套机制，用于实现高效的批处理。

• 用于调度和恢复的回溯法：由 Microsoft Dryad 引入，现在几乎用于所有批处理器；
• 用于散列和排序的特殊内存数据结构：可以在需要时，将一部分数据从内存溢出到硬盘上；
• 优化器：尽可能地缩短生成结果的时间。

流与批处理机制

两套机制分别对应各自的API（DataStream API 和 DataSet API）；在创建 Flink 作业时，并不能通过将两者混合在一起来同时 利用 Flink 的所有功能。

Flink支持两种关系型的API，Table APl和sQL。这两个API都是批处理和流处理统一的APl，这意味着在无边界的实时数据流和有边界的历史记录数据流上，关系型API会以相同的语义执行查询，并产生相同的结果。

• Table API / SQL 正在以流批统一的方式成为分析型用例的主要 API。
• DataStream API 是数据驱动应用程序和数据管道的主要API。

二、Flink的Time与Window

时间背景

在流处理器编程中,对于时间的处理是非常关键的。比如计数的例子，事件流数据（例如服务器日志数据、网页点击数据和交易数据）不断产生，我们需要用key将事件分组，并且每隔一段时间就针对每一个key对应的事件计数。这就是我们熟知的“大数据”应流处理中的时间分类

在数据流处理过程中，我们经常使用系统处理时间即: processing time作为某个事件的时间，而实际上系统时间processing time是我们强加给事件的时间，由于网络延迟等原因并不能较好的反应事件之间发生的先后顺序。

在实际场景中，每个事件的时间可以分为三种:

• event time，即事件发生时的时间;
• ingestion time，即事件到达流处理系统的时间;
• processing time，即事件被系统处理的时间。

三种时间示例

例如，一条日志进入Flink的时间为2019-11-1210:00:00.123，到达window的系统时间为2019-11-1210:00:01.234，日志的内容如下:

2019-11-0218:37:15.624 INFO Fail over to rm2

2019-11-0218:37:15.624是Event Time;

2019-11-1210:00:00.123是Ingestion Time;

2019-11-1210:00:01.234是Processing Time;

三种时间的区别

实际情况中事件真正发生的先后顺序与系统处理时间存在一定的差异，这些差异主要由网络延迟、处理时间的长短等造成。如图所示:

横坐标代表Event time，纵坐标代表processing time。理想情况下,eventtime和processing time构成的坐标应该形成一条倾斜角为45度的线。但实际应用过程中，processing time要落后与eventtime，造成事件到来的先后顺序不一致。

Flink支持的时间语义

Processing Time是指事件数据被Operator处理时所在机器的系统时间，它提供了最好的性能和最低的延迟。

Event Time是指在数据产生时该设备上对应的时间，这个时间在进入Flink之前已经存在于数据记录中了。

Ingestion Time指的是事件数据进入到Flink的时间。

Window概述

流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而Window是一种切割无限数据为有限块进行处理的手段。Window是无限数据流处理的核心，它将一个无限的stream拆分成有限大小的"buckets"桶，我们可以在这些桶上做计算操作。

Window类型

Window根据应用类型可以分成两类:

• CountWindow:数据驱动，按照指定的数据条数生成一个Window，与时间无关。
• TimeWindow:时间驱动，按照时间生成Window。

Apache Flink是一个天然支持无限流数据处理的分布式计算框架，在Flink中 Window可以将无限流切分成有限流。Flink中 Window可以是Time Window，也可以是Count Window。

TimeWindow分类

TimeWindow可以根据窗口实现原理的不同分成三类:滚动窗口（Tumbling Window ) .滑动窗口( Sliding Window)和会话窗口( Session Window)。

滚动窗口

将数据依据固定的窗口长度对数据进行切片。特点:时间对齐，窗口长度固定，没有重叠。

适用场景:适合做Bl统计等（做每个时间段的聚合计算）。

举一个例子，假设要对传感器输出的数值求和。一分钟滚动窗口收集最近一分钟的数值，并在一分钟结束时输出总和，如下图所示。

滑动窗口

滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。特点∶时间对齐，窗口长度固定，有重叠。

适用场景:对最近一个时间段内的统计（求某接口最近5min的失败率来决定是否要报警）。

示例:一分钟滑动窗口计算最近一分钟的数值总和，但每半分钟滑动一次并输出结果，如下图所示。

会话窗口

会话窗口由一系列事件组合一个指定时间长度的timeout间隙组成，类似于web应用的session，也就是一段时间没有接收到新数据就会生成新的窗口。特点:时间无对齐。

代码定义

在Flink中，一分钟滚动窗口的定义如下:

stream.timeWindow(Time.minutes(1));

在Flink中，每半分钟(即30秒）滑动一次的一分钟滑动窗口，如下所示:

stream.timeWindow(Time.minutes(1),Time.seconds(30));

三、Flink的Watermark

乱序问题

流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的，虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、分布式等原因，导致乱序的产生，所谓乱序，就是指Flink接收到的事件的先后顺序不是严格按照事件的Event Time顺序排列的。

此时出现一个问题，一旦出现乱序，如果只根据eventTime决定window的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了，这个特别的机制，就是Watermark。

乱序示例

例子:某App会记录用户的所有点击行为，并回传日志（在网络不好的情况下，先保存在本地，延后回传）。A用户在11:02对App进行操作，B用户在11:03对App进行操作，但是A用户的网络不太稳定，回传日志延迟了，导致我们在服务端先接受到B用户11:03的消息，然后再接受到A用户11:02的消息，消息乱序了。

水位线（Watermark）

对于无穷数据集，我们缺乏一种有效的方式来判断数据完整性，因此就有了Watermark,它是建立在事件时间上的一个概念，用来刻画数据流的完整性。如果按照处理时间来衡量事件，一切都是有序的、完美的，自然而然也就不需要Watermark了。换句话说事件时间带来了乱序的问题，而Watermark就是用来解决乱序问题。所谓的乱序，其实就是有事件延迟了，对于延迟的元素，我们不可能无限期的等下去，必须要有一种机制来保证一个特定的时间后，必须触发Window进行计算。这个特别的机制,就是Watermark,它告诉了算子延迟到达的消息不应该再被接收。

• Watermark是一种衡量Event Time进展的机制。
  
• Watermark是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark机制结合window来实现。
  
• 数据流中的Watermark用于表示timestamp小于Watermark的数据，都已经到达了，因此，window的执行也是由Watermark触发的。
  
• Watermark可以理解成一个延迟触发机制，我们可以设置Watermark的延时时长t，每次系统会校验已经到达的数据中最大的maxEventTime，然后认定eventTime小于maxEventTime - t的所有数据都已经到达，如果有窗口的停止时间等于maxEventTime – t，那么这个窗口被触发执行。
  
• watermark 用来让程序自己平衡延迟和结果正确性

Watermark的原理

Flink怎么保证基于event-time的窗口在销毁的时候，已经处理完了所有的数据呢?

这就是watermark的功能所在。watermark会携带一个单调递增的时间戳t，Watermark(t)表示所有时间戳不大于t的数据都已经到来了，未来小于等于t的数据不会再来，因此可以放心地触发和销毁窗口了。

当Flink,接收到数据时，会按照一定的规则去生成Watermark，这条Watermark就等于当前所有到达数据中的maExertT me"-延N时长，也就定说，Watermark是基于数据携带的时间戳生成的，一旦Watermark比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。由于eventtime是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。

上图中，我们设置的允许最大延迟到达时间为2s，所以时间戳为7s的事件对应的Watermark 是 5s，时间戳为12s的事件的Watermark是10s，如果我们的窗口是1s-5s，窗口2是6s~-10s，那么时间戳为7s的事件到达时的Matermarker.恰好触发窗口1，时间戳为 12s的事件到达时的Watermark恰好触发窗口2。

Watermark就是触发前一窗口的“关窗时间”，一旦触发关门那么以当前时刻为准在窗口范围内的所有所有数据都会收入窗中。只要没有达到水位那么不管现实中的时间推进了多久都不会触发关窗。

延迟的数据

Watermark能够应对乱序的数据，但是真实世界中没法得到一个完美的 Watermark数值。要么没法获取到，要么耗费太大，因此实际工作中会近似 Watermark(t)之后，还有较小的概率接受到时间戳t之前的数据，在Flink中将这些数据定义为“late elements”，同样可以在Window中指定允许延迟的最大时间（默认为О)，可以使用下面的代码进行设置:

延迟数据处理机制

延迟事件是乱序事件的特例，和一般乱序事件不同的是它们的乱序程度超出了水位线( Watermark)的预计，导致窗口在它们到达之前已经关闭。

延迟事件出现时窗口已经关闭并产出了计算结果，对于此种情况处理的方法有3种:

• 重新激活已经关闭的窗口并重新计算以修正结果。
• 将延迟事件收集起来另外处理。
• 将延迟事件视为错误消息并丢弃。

Flink默认的处理方式是第3种直接丢弃，其他两种方式分别使用Side Output和AllowedLateness。

Side Output机制

Side Output机制可以将延迟事件单独放入一个数据流分支，这会作为Window计算结果的副产品，以便用户获取并对其进行特殊处理。

side Output获取延迟数据：

设置allowedLateness之后，迟来的数据同样可以触发窗口，进行输出，利用Flink的sideoutput机制，可以获取到这些延迟的数据，使用方式如下:

Allowed Lateness机制

Allowed Lateness机制允许用户设置一个允许的最大延迟时长。Flink会在窗口关闭后一直保存窗口的状态直至超过允许延迟时长，这期间的延迟事件不会被丢弃，而是默认会触发窗口重新计算。因为保存窗口状态需要额外内存，并且如果窗口计算使用了ProcessWindowFunction APl还可能使得每个延迟事件触发一次窗口的全量计算，代价比较大，所以允许延迟时长不宜设得太长，延迟事件也不宜过多。

四、Flink的容错

Flink容错机制

为了保证程序的容错恢复以及程序启动时其状态恢复，Flink任务都会开启Checkpoint或者触发Savepoint进行状态保存。

• Checkpoint机制。这种机制保证了实时程序运行时，即使突然遇到异常也能够进行自我恢复。Checkpoint对于用户层面，是透明的，用户会感觉不到Checkpoint过程的存在。
• Savepoint机制。是在某个时间点程序状态全局镜像，以后程序在进行升级，或者修改并发度等情况，还能从保存的状态位继续启动恢复。Savepoint可以看做是Checkpoint在特定时期的一个状态快照。

Checkpoint

Flink 如何保证exactly-once呢?它使用一种被称为“检查点( Checkpoint )”的特性，在出现故障时将系统重置回正确状态。Flink状态保存主要依靠Checkpoint机制,Checkpoint会定时制作分布式快照，对程序中的状态进行备份。

Checkpoint检查点机制

Flink中基于异步轻量级的分布式快照技术提供了Checkpoints容错机制，分布式快照可以将同一时间点Task/Operator的状态数据全局统一快照处理。Flink会在输入的数据集上间隔性地生成checkpoint barrier，通过棚栏( barrier)将间隔时间段内的数据划分到相应的checkpoint中。当应用出现异常时，Operator就能够从上一次快照中恢复所有算子之前的状态，从而保证数据的一致性。

对于状态占用空间比较小的应用，快照产生过程非常轻量，高频率创建且对Flink任务性能影响相对较小。Checkpoint过程中状态数据一般被保存在一个可配置的环境中，通常是在JobManager节点或HDFS上。

Checkpoint配置

默认情况下Flink不开启检查点，用户需要在程序中通过调用enableCheckpointing(n)方法配置和开启检查点，其中n为检查点执行的时间间隔，单位为毫秒。

exactly-once和at-least-once语义选择

exactly-once:保证端到端数据一致性，数据要求高，不允许出现数据丢失和数据重复，Flink的性能也相对较弱;

at-least-once:时延和吞吐量要求非常高但对数据的一致性要求不高的场景。

Flink默认使用exactly-once模式，可以通过setCheckpointingMode()方法来设定语义模式。

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)

Checkpoint超时时间

指定每次Checkpoint执行过程中的上限时间范围，一旦Checkpoint执行时间超过该阈值，Flink将会中断Checkpoint过程,并按照超时处理。

该指标可以通过setCheckpointTimeout方法设定，默认10分钟。

env.getCheckpointConfig().setCheckpointingTimeout(60000)

检查点之间最小时间间隔

设定两个Checkpoint之间的最小时间间隔，防止出现状态数据过大而导致Checkpoint执行时间过长，从而导致Checkpoint积压过多，最终Flink应用密集地触发Checkpoint操作，会占用大量计算资源而影响到整个应用的性能。

env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500)

最大并行执行的检查点数量