说明：本文为《Flink大数据项目实战》学习笔记，想通过视频系统学习Flink这个最火爆的大数据计算框架的同学，推荐学习CSDN官网课程：
Flink大数据项目实战：http://t.cn/ExrHPl9
（2）Flink初探
1.快速生成Flink项目
1.推荐开发工具
idea+maven+git
2.推荐开发语言
Java或者Scala
https://ci.apache.org/projects/flink/flink-docs-release-1.6/quickstart/java_api_quickstart.html
3.Flink项目构建步骤
1）通过maven构建Flink项目
这里我们选择构建1.6.2版本的Flink项目，打开终端输入如下命令：
mvn archetype:generate -DarchetypeGroupId=org.apache.flink -DarchetypeArtifactId=flink-quickstart-java -DarchetypeVersion=1.6.2
项目构建过程中需要输入groupId，artifactId，version和package
然后输入y确认
然后显示Maven项目构建成功
2）打开IDEA导入Flink 构建的maven项目
打开IDEA开发工具，点击open选项
选择刚刚创建的Flink项目
IDEA打开Flink项目
2. Flink Batch版WordCount
新建一个batch package
打开github Flink源码，将批处理WordCount代码copy到batch包下。
右键选择run，运行Flink批处理WordCount，运行结果如下所示：
3. Flink Stream版WordCount
同样，流处理我们也单独创建一个包stream
打开github Flink源码，将流处理WordCount代码copy到stream包下。
打开流处理WordCount代码：
右键选择run，运行Flink流处理WordCount，运行结果如下所示：
（3）Flink核心概念与编程模型
1. Flink分层架构
1.1 Flink生态之核心组件栈
大家回顾一下Flink生态圈中的核心组件栈即可，前面已经详细讲过，这里就不再赘叙。
1.2 Flink分层架构
Flink一共分为四个层级，具体如下图所示：
Flink最下面的一层API为Stateful Stream Processing，它是Flink最底层的API，控制更灵活但一般很少使用。然后上面一层就是Flink Core（核心）API，它包含DataStream和DataSet API，应用层的用户经常使用 Core API。然后再上面一层就是 Table API，它相当于在Core API中可以定义数据的Table结构，可以做table操作。最上面一层就是SQL 操作，用户可以直接使用SQL语句对数据处理，更简单更方便。
注意：越底层的API越灵活，但越复杂。越上层的API越轻便，但灵活性差。
[if !supportLists]1. [endif]Stateful Stream Processing
a）它位于最底层，是Core API 的底层实现。
b）它是嵌入到Stream流里面的处理函数（processFunction）。
c）当Core API满足不了用户需求，可以利用低阶API构建一些新的组件或者算子。
d）它虽然灵活性高，但开发比较复杂，需要具备一定的编码能力。
[if !supportLists]2. [endif]Core API
[if !supportLists]a） [endif]DataSet API是批处理API，处理有限的数据集。
[if !supportLists]b） [endif]DataStream API是流处理API，处理无限的数据集。
[if !supportLists]3. [endif]Table API & SQL
a）SQL 构建在Table 之上，都需要构建Table 环境。
b）不同的类型的Table 构建不同的Table 环境中。
c）Table 可以与DataStream或者DataSet进行相互转换。
d）Streaming SQL不同于存储的SQL，最终会转化为流式执行计划。
1.3Flink DataFlow
Flink DataFlow基本套路：先创建Data Source读取数据，然后对数据进行转化操作，然后创建DataSink对数据输出。
结合代码和示意图理解DataFlow
Flink DataFlow基本套路如下所示：
步骤1：构建计算环境(决定采用哪种计算执行方式)
步骤2：创建Source(可以多个数据源)
步骤3：对数据进行不同方式的转换(提供了丰富的算子)
步骤4：对结果的数据进行Sink(可以输出到多个地方)
并行化DataFlow
从上图可以看出Source的并行度为2，它们可以并行运行在不同的节点上。Map的并行度也为2，source读取数据后做Stream Partition操作，source1将数据交给map1，source2将数据交给map2。keyBy（或者window等）的并行度为2，map处理后的数据需要经过shuffle操作，然后交给keyBy进行分组统计。Sink的并行度为1，keyBy最后分组统计后的数据交给sink，将数据进行输出操作。
算子间数据传递模式
从上图可以看出，Flink算子间的数据传递模式大概分为两种：
1.One-to-one streams：保持元素的分区和顺序，比如数据做map操作。
2.Redistributing streams： 它会改变流的分区，重新分区策略取决于使用的算子
keyBy() (re-partitions by hashing the key)：根据hash key对数据重新分区。
broadcast()：即为广播操作，比如map1有100条数据，发送给keyBy1是100条数据，发给keyBy2也是100条数据。
rebalance() (which re-partitions randomly)：即随机打散，数据随机分区发送给下游操作。
2. Windows
前面我们已经了解了Flink的Stream流处理和Batch批处理，那么我们这里讲的Windows操作是对一段数据进行操作，它可以按照固定数据量进行Windows操作，也可以按照固定时间进行windows操作，它是Stream 流处理所特有的窗口操作。
Flink Windows操作的类型大概分为以下几类：
[if !supportLists]1. [endif]Count Windows
顾名思义，是按照Events的数量进行操作，比如每3个Event做一次windows操作。
[if !supportLists]2. [endif]Time Windows
基于时间长度进行Windows操作
[if !supportLists]a） [endif]Tumbling Windows：即翻滚窗口，不会重叠，比如每隔3s操作一次。
[if !supportLists]b） [endif]Sliding Windows：即滑动窗口，有重叠，比如窗口大小为3s，每次向前滑动1s。
[if !supportLists]c） [endif]Session Windows：类似于Web编程里的Session，以不活动间隙作为窗口进行操作，比如每10s内没有活动，就会做一次Windows操作。
[if !supportLists]3. [endif]自定义Windows
当Flink内置的windows不能满足用户的需求，我们可以自定义Windows操作。
3.各种Time
从上图可以看出Flink中的Time大致分为以下三类：
1.Event Time：Event 真正产生的时间，我们称之为Event Time。
2.Ingestion Time：Event 事件被Source拿到，进入Flink处理引擎的时间，我们称之为Ingestion Time。
3.Window Processing Time：Event事件被Flink 处理（比如做windows操作）时的时间，我们称之为Window Processing Time。
4.Stateful Operations
什么是状态？
state一般指一个具体的task/operator的状态，比如当前处理那些数据，数据处理的进度等等。
Flink state操作状态分为两类：
1.Operator State
Operator State跟一个特定operator的一个并发实例绑定，整个operator只对应一个state。
2.Keyed State
基于KeyedStream上的状态。这个状态是跟特定的key绑定的，对KeyedStream流上的每一个key，可能都对应一个state。
Flink每个操作状态又分为两类：
Keyed State和Operator State可以以两种形式存在：原始状态和托管状态（ Flink框架管理的状态）。
1.原始状态：比如一个字符串或者数组，它需要序列化，保存到内存或磁盘，或者外部存储中，这就是它的原始状态。
2.托管状态：比如数据放在Hash表中，或者放在HDFS中，或者放在rocksdb中，这种就是托管状态。当需要处理数据的时候，从托管状态中读取出来，还原成原始状态，甚至变量和集合，然后再进行处理。
5.Checkpoints(备份)
什么是checkpoint？
所谓checkpoint，就是在某一时刻，将所有task的状态做一个快照(snapshot)，然后存储到State Backend（比如hdfs）。checkpoint拥有轻量级容错机制，可以保证exactly-once 语义，用于内部失败的恢复(比如当应用挂了，它可以自动恢复从上次的进度接着执行)。
checkpoint基本原理：通过往source 注入barrier（可以理解为特殊的Event），barrier作为checkpoint的标志，它会自动做checkpoint无需人工干预。
6.Savepoint
savepoint是流处理过程中的状态历史版本，它具有可以replay的功能。用于外部恢复，当Flink应用重启和升级，它会做一个先做一个savepoint，下次应用启动可以接着上次进度执行。
savepoint两种触发方式：
1.Cancel with savepoint
2.手动主动触发
savepoint可以理解为是一种特殊的checkpoint，savepoint就是指向checkpoint的一个指针，需要手动触发，而且不会过期，不会被覆盖，除非手动删除。正常情况下的线上环境是不需要设置savepoint的。除非对job或集群做出重大改动的时候，需要进行测试运行。
（4）Flink Runtime
1. Flink运行时架构
1.1Flink架构
Flink运行时架构主要包含几个部分：Client、JobManager(master节点)和TaskManger(slave节点)。
Client：Flink 作业在哪台机器上面提交，那么当前机器称之为Client。用户开发的Program 代码，它会构建出DataFlow graph，然后通过Client提交给JobManager。
JobManager：是主（master）节点，相当于YARN里面的REsourceManager，生成环境中一般可以做HA 高可用。JobManager会将任务进行拆分，调度到TaskManager上面执行。
TaskManager：是从节点（slave），TaskManager才是真正实现task的部分。
Client提交作业到JobManager，就需要跟JobManager进行通信，它使用Akka框架或者库进行通信，另外Client与JobManager进行数据交互，使用的是Netty框架。Akka通信基于Actor System，Client可以向JobManager发送指令，比如Submit job或者Cancel /update job。JobManager也可以反馈信息给Client，比如status updates，Statistics和results。
Client提交给JobManager的是一个Job，然后JobManager将Job拆分成task，提交给TaskManager（worker）。JobManager与TaskManager也是基于Akka进行通信，JobManager发送指令，比如Deploy/Stop/Cancel Tasks或者触发Checkpoint，反过来TaskManager也会跟JobManager通信返回Task Status，Heartbeat（心跳），Statistics等。另外TaskManager之间的数据通过网络进行传输，比如Data Stream做一些算子的操作，数据往往需要在TaskManager之间做数据传输。
1.2. TaskManger Slot
TaskManager是进程，他下面运行的task（整个Flink应用是Job，Job可以拆分成很多个task）是线程，每个task/subtask(线程)下可运行一个或者多个operator，即OperatorChain。Task是class，抽象的，subtask是Object(类比学习)，具体的。
一个TaskManager通过Slot(任务槽)来控制它上面可以接受多少个task，比如一个TaskManager划分了3个Task Slot(仅限内存托管，目前CPU未做隔离)，它只能接受3个task。Slot均分TaskManager所托管的内存，比如一个TaskManager有6G内存，那么每个Slot分配2G。
同一个TaskManager中的task共享TCP连接(通过多路复用)和心跳消息。它们还可以共享数据集和数据结构，从而减少每个任务的开销。一个TaskManager有N个槽位只能接受N个Task吗？不是，后面会讲共享槽位。
1.3. OperatorChain && Task
为了更高效地分布式执行，Flink会尽可能地将operator的subtask链接（chain）在一起形成task。以wordcount为例，解析不同视图下的数据流，如下图所示。
数据流（逻辑视图）
创建Source（并行度设置为1）读取数据源，数据经过FlatMap（并行度设置为2）做转换操作，然后数据经过Key Agg（并行度设置为2）做聚合操作，最后数据经过Sink（并行度设置为2）将数据输出。
数据流（并行化视图）
并行度为1的Source读取数据源，然后FlatMap并行度为2读取数据源进行转化操作，然后数据经过Shuffle交给并行度为2的Key Agg进行聚合操作，然后并行度为2的Sink将数据输出，未优化前的task总和为7。
数据流（优化后视图）
并行度为1的Source读取数据源，然后FlatMap并行度为2读取数据源进行转化操作，然后数据经过Shuffle交给Key Agg进行聚合操作，此时Key Agg和Sink操作合并为一个task（注意：将KeyAgg和Sink两个operator进行了合并，因为这两个合并后并不会改变整体的拓扑结构），它们一起的并行度为2，数据经过Key Agg和Sink之后将数据输出，优化后的task总和为5.
1.4. OperatorChain的优点和组成条件
OperatorChain的优点
1.减少线程切换
2.减少序列化与反序列化
3.减少数据在缓冲区的交换
4.减少延迟并且提高吞吐能力
OperatorChain组成条件
1.没有禁用Chain
2.上下游算子并行度一致 。
3.下游算子的入度为1(也就是说下游节点没有来自其他节点的输入)。
4.上下游算子在同一个slot group(后面紧跟着就会讲如何通过slot group先分配到同一个solt，然后才能chain) 。
5.下游节点的 chain 策略为 ALWAYS（可以与上下游链接，map、flatmap、filter等默认是ALWAYS）。
6.上游节点的 chain 策略为 ALWAYS 或 HEAD（只能与下游链接，不能与上游链接，Source默认是HEAD）。
7.上下游算子之间没有数据shuffle (数据分区方式是 forward)。
1.5.编程改变OperatorChain行为
Operator chain的行为可以通过编程API中进行指定，可以通过在DataStream的operator后面（如someStream.map(..))调用startNewChain()来指示从该operator开始一个新的chain（与前面截断，不会被chain到前面）。可以调用disableChaining()来指示该operator不参与chaining（不会与前后的operator chain一起）。可以通过调用StreamExecutionEnvironment.disableOperatorChaining()来全局禁用chaining。可以设置Slot group，例如someStream.filter(...).slotSharingGroup(“name”)。可以通过调整并行度，来调整Operator chain。
2. Slot分配与共享
2.1共享Slot
默认情况下，Flink允许subtasks共享slot，条件是它们都来自同一个Job的不同task的subtask。结果可能一个slot持有该job的整个pipeline。
允许slot共享有以下两点好处：
1.Flink集群需要的任务槽与作业中使用的最高并行度正好相同(前提，保持默认SlotSharingGroup)。也就是说我们不需要再去计算一个程序总共会起多少个task了。
2.更容易获得更充分的资源利用。如果没有slot共享，那么非密集型操作source/flatmap就会占用同密集型操作 keyAggregation/sink 一样多的资源。如果有slot共享，将task的2个并行度增加到6个，能充分利用slot资源，同时保证每个TaskManager能平均分配到重的subtasks。
2.2共享Slot实例
将WordCount的并行度从之前的2个增加到6个（Source并行度仍为1），并开启slot共享（所有operator都在default共享组），将得到如上图所示的slot分布图。
首先，我们不用去计算这个job会其多少个task，总之该任务最终会占用6个slots（最高并行度为6）。其次，我们可以看到密集型操作 keyAggregation/sink 被平均地分配到各个 TaskManager。
2.3 SlotSharingGroup(soft)
SlotSharingGroup是Flink中用来实现slot共享的类，它尽可能地让subtasks共享一个slot。
保证同一个group的并行度相同的sub-tasks 共享同一个slots。算子的默认group为default(即默认一个job下的subtask都可以共享一个slot)
为了防止不合理的共享，用户也能通过API来强制指定operator的共享组，比如：someStream.filter(...).slotSharingGroup("group1");就强制指定了filter的slot共享组为group1。怎么确定一个未做SlotSharingGroup设置算子的SlotSharingGroup什么呢(根据上游算子的group 和自身是否设置group共同确定)。适当设置可以减少每个slot运行的线程数，从而整体上减少机器的负载。
2.4 CoLocationGroup(强制)
CoLocationGroup可以保证所有的并行度相同的sub-tasks运行在同一个slot，主要用于迭代流(训练机器学习模型)。
3. Slot & parallelism的关系
3.1 Slots && parallelism
3.2如何计算Slot
如何计算一个应用需要多少slot？
如果不设置SlotSharingGroup，那么需要的Slot数为应用的最大并行度数。如果设置了SlotSharingGroup，那么需要的Slot数为所有SlotSharingGroup中的最大并行度之和。比如已经强制指定了map的slot共享组为test，那么map和map下游的组为test，map的上游source的组为默认的default，此时default组中最大并行度为10，test组中最大并行度为20，那么需要的Slot=10+20=30。
