Flink入门到实践-京东云开发者社区

### 1 导言
通过本文可以快速的了解Flink开发的模式和开发的时候最常见的概念，算子、位点提交、slot和并行度都是开发工作中必须要了解的知识；我们已对Flink相关的功能进行了封装，开发者不需要关注数据如何接入、如何输出；只需要关注自己的业务逻辑即可，有需求的可联系我们进行探讨。
### 1.1 Flink介绍
- Flink是一个支持分布式、高性能、高可用性、高可靠性的流式处理框架
高吞吐、低延迟、高性能
- 批流统一:Runtime和SQL层批流统一，提供高吞吐低延时就算能力和更强大的SQL支持。
- 生态兼容：与Hadoop Yarn/Apache Mesos等集成，并且支持单机模式运行。
- 规模计算：作业可被分摊成若干个任务，分布在集群中并发执行。
- 高度灵活的窗口（Window）操作，支持基于time、count、session窗口
- 用Event Time 和 Water Mark 等先进技术解决消息乱序问题
- 具有Backpressure 以及系统自带监控功能
- 有状态,轻量级分布式快照（Snapshot）,增量checkpoint机制实现容错
- 自己的内存管理， 与Spark tungsten计划类比

#### 1.2 Flink storm比较

![](//img1.jcloudcs.com/developer.jdcloud.com/c6f2429a-f80d-44d1-bbe4-93007d23474a20210729135450.png)

### 2 消息投递方式
- At Most Once：至多一次，表示一条消息不管后续处理成功与否只会被消费处理一次，那么就存在数据丢失可能、
- At Least Once：至少一次，表示一条消息从消费到后续的处理成功，可能会发生多次
- Exactly Once：精确一次，表示一条消息从其消费到后续的处理成功，只会发生一次

### 3 通用接入模式

![](//img1.jcloudcs.com/developer.jdcloud.com/07c953b4-9565-41c3-a8be-34f95ba5dd4020210729135609.png)

**source：**数据输入源，基于本地接的source、基于文件的source、基于网络套接字的source、自定义source;自定义的通常有JDQ,JMQ

**transfer：**数据转换的各种操作（算子），将通过source接收的数据按照业务逻辑加工（将数据转换成我们想要的数据）

**sink：**接收器(输出)，将转换计算后的数据发送的地点；常见的有：写入文件、打印出来、写入socket、自定义sink;，自定义的sink有mysql,elasticsearch,redis,jdq,mq

### 4 算子介绍
![](//img1.jcloudcs.com/developer.jdcloud.com/9a77370e-a46a-4431-b20c-4c570fdd7eb320210729135657.png)

#### 4.1 fliter 
意思是进行筛选
将成绩小于400分的的同学信息过滤掉，只要总成绩大于400的

![](//img1.jcloudcs.com/developer.jdcloud.com/f4976783-f3ef-40f5-8c79-fb0dfd1e153920210729135720.png)

#### 4.2 map 
意思是对每个元素进行一定的变换，映射为另一个元素
将每个学生的总成绩加20分

![](//img1.jcloudcs.com/developer.jdcloud.com/d9c5a2cf-7989-4329-b1cb-d3e3750597d520210729135746.png)

#### 4.3 flatmap
意思是为将元素摊平，每个元素可以变为0个、1个、或者多个元素
将每个学生的每个学科的成绩信息打印出来

![](//img1.jcloudcs.com/developer.jdcloud.com/62393b06-0ffa-4b19-910e-ac0ff260b90b20210729135806.png)

#### 4.4 keyby
意思是根据指定的Key进行分区，是根据key的散列值进行分区
按照学科名称将课程信息按分区进行keyby

![](//img1.jcloudcs.com/developer.jdcloud.com/af64705b-acc7-4b2c-b707-fc0f8af60c1120210729135830.png)

#### 4.5 侧输出 
一个流有多个输出，侧输出结果流的数据类型不需要与主数据流的类型一致，不同侧输出流的类型也可以不同
侧输出类型定义：source为70的

![](//img1.jcloudcs.com/developer.jdcloud.com/7c8f2f91-8a73-4173-9569-40650fdcc41c20210729135850.png)

侧输出处理：将source=70的数据输出到上面定义的outputTag中

![](//img1.jcloudcs.com/developer.jdcloud.com/3c0268da-9d4d-4da4-ac14-365307cae5bc20210729135909.png)

侧输出数据流获取

![](//img1.jcloudcs.com/developer.jdcloud.com/b0169248-3424-4d01-8a86-5de739d37dba20210729135929.png)

#### 4.6 窗口
window可以帮助我们将无限流拆分成一定大小的模块；在Flink中，window的使用如下：
```java
来获取侧输出流,当窗口（window）f// Keyed Window
stream
       .keyBy(...)               <-  按照一个Key进行分组
       .window(...)              <-  将数据流中的元素分配到相应的窗口中
      [.trigger(...)]            <-  指定触发器Trigger（可选）
      [.evictor(...)]            <-  指定清除器Evictor(可选)
       .reduce/aggregate/process()      <-  窗口处理函数Window Function
// Non-Keyed Window
stream
       .windowAll(...)           <-  不分组，将数据流中的所有元素分配到相应的窗口中
      [.tigger(...)]            <-  指定触发器Trigger（可选）
      [.evictor(...)]            <-  指定清除器Evictor(可选)
       .reduce/aggregate/process()      <-  窗口处理函数Window Function
```
Flink窗口的骨架结构中有两个必须的操作：
- 使用窗口分配器（WindowAssigner）将数据流中的元素分配到对应的窗口。
- 当满足窗口触发条件后，对窗口内的数据使用窗口处理函数（Window Function）进行处理，常用的Window Function有reduce、aggregate、process。

##### 4.6.1 滚动窗口
窗口之间之间不重叠，且窗口长度是固定的，可以用TumblingEventTimeWindows和TumblingProcessingTimeWindows创建一个基于Event Time或Processing Time的滚动时间窗口

![](//img1.jcloudcs.com/developer.jdcloud.com/1ddb1923-e47b-4399-b23f-af1f7874607520210729140646.png)

##### 4.6.2 滑动窗口
以一个步长(slide)不断向前滑动，窗口的长度固定,可以用SlidingEventTimeWindows和SlidingProcessingTimeWindows来创建滑动窗口

![](//img1.jcloudcs.com/developer.jdcloud.com/a7000067-4fc3-46bf-b2db-b1b0d56ddd0320210729140706.png)

##### 4.6.3 会话窗口
根据Session gap切分不同的窗口，当一个窗口在大于Session gap的时间内没有接收到新数据时，窗口将关闭

![](//img1.jcloudcs.com/developer.jdcloud.com/425e1017-91e3-45e3-a5e3-ccddbb89b83620210729140740.png)

### 5 checkpoint（位点提交）
再可靠的框架也有出现各种原因导致整个任务失败的可能性，Flink提供的位点可以最大限度的让我们在任务异常时恢复任务，位点提交的过程如下：

![](//img1.jcloudcs.com/developer.jdcloud.com/27cfe7a6-bde9-4d54-b8d1-303949bf3a3820210729140802.png)

1. JobManager向所有的source 发送checktpointTrigger,Source Task会在数据流中安插CheckPoint barrier
1. Source Task在接收到barrier的时候自身做快照，并保存到状态后端
1. Source Task在完成自身快照之后将barrier跟数据流一块往下游发送
1. 当下游的Operator实例接收到CheckPoint barrier后，对自身做快照
1. 当所有task完成快照之后，将地址信息通知JobManger
1. 当JobManager收到所有完成task的反馈信息之后，将这些备份数据的地址写入可持久化存储中

位点提交的时候会有barrrier对齐和不对齐的情况： 一旦Operator从输入流接收到CheckPoint barrier n，它就不能处理来自该流的任何数据记录，直到它从其他所有输入接收到barrier n为止。否则，它会混合属于快照n的记录和属于快照n + 1的记录；接收到barrier n的流暂时被搁置。从这些流接收的记录不会被处理，而是放入输入缓冲区。一旦最后所有输入流都接收到barrier n，Operator就会把缓冲区中pending 的输出数据发出去，然后把CheckPoint barrier n接着往下游发送

### 6 parallelism和Slot
#### 6.1 parallelism 
指 taskmanager 实际使用的并发能力

![](//img1.jcloudcs.com/developer.jdcloud.com/214a4c5c-52e7-48e3-89b9-e3de4a5babb320210729140856.png)

parallelism.default:1
运行程序默认的并行度为 1，9 个 TaskSlot 只用了 1 个，有 8 个空闲。设置合适的并行度才能提高效率。

#### 6.2 slot
taskmanager 的并发执行能力
slot 的数量通常与每个 TaskManager 的可用 CPU 内核数成比例。一般情况下你的 slot 数是,slot对Taskmanager内存进行平均分配，每个slot内存相同，加起来和等于taskManager可用内存，但是仅仅对内存做了隔离，并没有对CPU进行隔离。你每个 TaskManager 的 cpu 的核数

![](//img1.jcloudcs.com/developer.jdcloud.com/69c3c383-cefa-4d42-8497-35e23e5b2b9820210729140924.png)

taskmanager.numberOfTaskSlots:3
每一个 taskmanager 中的分配 3 个 TaskSlot, 3 个 taskmanager 一共有 9 个 TaskSlot。
#### 6.3 SlotSharingGroup

Flink 允许subtasks共享slot，条件是它们都来自同一个Job的不同task的subtask.

通过调整 task slot 的数量，用户可以定义task之间是如何相互隔离的。每个 TaskManager 有一个slot，也就意味着每个task运行在独立的 JVM 中。每个 TaskManager 有多个slot的话，也就是说多个task运行在同一个JVM中。而在同一个JVM进程中的task，可以共享TCP连接（基于多路复用)和减少数据的网络传输。也能共享一些数据结构，一定程度上减少了每个task的消耗。

### 7 运行Flink程序流程
#### 7.1 申请JRC资源
Flink任务运行之前得先申请运行的机器资源，可以在http://bdp.jd.com/jrdw/jrctask2/flink/management/titleMenuNoLeft.html?url=%2Fflink%2FclusterList上按如下方式申请；

![](//img1.jcloudcs.com/developer.jdcloud.com/36a93dc3-50ac-4038-b33d-3e7d1f3da21d20210729141004.png)
<center>集群管理页面</center>

![](//img1.jcloudcs.com/developer.jdcloud.com/fb0da07c-85e9-469c-b651-19ef6dde40ad20210729141047.png)
<center>申请机器资源页面</center>

#### 7.2 集群创建
在申请的资源审批通过之后，可基于申请的资源创建所需的运行集群，可以在集群创建页面配置集群的配置参数；其中集群创建页面的配置项所属应用和我们刚才申请机器资源页面的服务目录保持一致；引擎高级配置可以手动配置一些特殊配置项；比如使用JSF 的时候类加载方式以及位点提交方式配置等。如果Flink程序中有读写大数据平台的功能，需要打开读写大数据平台集市选项，并配置大数据平台集市信息。

![](//img1.jcloudcs.com/developer.jdcloud.com/416b0cb0-cff5-43f9-8ba0-ab243544fcf920210729141132.png)

![](//img1.jcloudcs.com/developer.jdcloud.com/5e184a62-4d11-4135-b0ba-d266a01fddd920210729141140.png)
<center>Flink集群创建页面</center>

#### 7.3 Flink运行任务构建
构建Flink任务的时候，建议选择Git，避免用Jar包的时候，多人协同开发因本地代码提交不及时的造成的非必要问题；其中的默认并行度是整个程序的并行度，如果不在程序中单独设置每个算子的并行度的话，source 到 operator到sink并行度都一样，构建完成之后，点击下一步可参看整个flink任务的运行拓扑，最后对审批完成的任务进行发布操作。

![](//img1.jcloudcs.com/developer.jdcloud.com/a803cd40-3eb7-498c-88e2-e5e4e0d1065220210729141215.png)

------------
###### 自猿其说Tech-JDL京东物流技术发展部
###### 作者：客户服务技术部 柳栋栋