Flink保姆级教程，超全五万字，学习与面试收藏这一篇就够了
本文目录：
一、Flink 简介
二、Flink 部署及启动
三、Flink 运行架构
四、Flink 算子大全
五、流处理中的 Time 与 Window
六、Flink 状态管理
七、Flink 容错
八、Flink SQL
九、Flink CEP
十、Flink CDC
十一、基于 Flink 构建全场景实时数仓
十二、Flink 大厂面试题
Flink 涉及的知识点如下图所示，本文将逐一讲解：
本文档参考了关于 Flink 的官网及其他众多资料整理而成，为了整洁的排版及舒适的阅读，对于模糊不清晰的图片及黑白图片进行重新绘制成了高清彩图。
一、Flink 简介
1. Flink 发展
这几年大数据的飞速发展，出现了很多热门的开源社区，其中著名的有 Hadoop、Storm，以及后来的 Spark，他们都有着各自专注的应用场景。Spark 掀开了内存计算的先河，也以内存为赌注，赢得了内存计算的飞速发展。Spark 的火热或多或少的掩盖了其他分布式计算的系统身影。就像 Flink，也就在这个时候默默的发展着。
在国外一些社区，有很多人将大数据的计算引擎分成了 4 代，当然，也有很多人不会认同。我们先姑且这么认为和讨论。
首先第一代的计算引擎，无疑就是 Hadoop 承载的 MapReduce。这里大家应该都不会对 MapReduce 陌生，它将计算分为两个阶段，分别为 Map 和 Reduce。对于上层应用来说，就不得不想方设法去拆分算法，甚至于不得不在上层应用实现多个 Job 的串联，以完成一个完整的算法，例如迭代计算。
由于这样的弊端，催生了支持 DAG 框架的产生。因此，支持 DAG 的框架被划分为第二代计算引擎。如 Tez 以及更上层的 Oozie。这里我们不去细究各种 DAG 实现之间的区别，不过对于当时的 Tez 和 Oozie 来说，大多还是批处理的任务。
接下来就是以 Spark 为代表的第三代的计算引擎。第三代计算引擎的特点主要是 Job 内部的 DAG 支持（不跨越 Job），以及强调的实时计算。在这里，很多人也会认为第三代计算引擎也能够很好的运行批处理的 Job。
随着第三代计算引擎的出现，促进了上层应用快速发展，例如各种迭代计算的性能以及对流计算和 SQL 等的支持。Flink 的诞生就被归在了第四代。这应该主要表现在 Flink 对流计算的支持，以及更一步的实时性上面。当然 Flink 也可以支持 Batch 的任务，以及 DAG 的运算。
总结：
第 1 代： Hadoop MapReduc 批处理 Mapper、Reducer 2；
第 2 代： DAG 框架（Oozie 、Tez），Tez + MapReduce 批处理 1 个 Tez = MR(1) + MR(2) + ... + MR(n) 相比 MR 效率有所提升；
第 3 代： Spark 批处理、流处理、SQL 高层 API 支持 自带 DAG 内存迭代计算、性能较之前大幅提；
第 4 代： Flink 批处理、流处理、SQL 高层 API 支持 自带 DAG 流式计算性能更高、可靠性更高。
2. 什么是 Flink
Flink 起源于 Stratosphere 项目，Stratosphere 是在 2010~2014 年由 3 所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目，2014 年 4 月 Stratosphere 的代码被复制并捐赠给了 Apache 软件基金会，参加这个孵化项目的初始成员是 Stratosphere 系统的核心开发人员，2014 年 12 月，Flink 一跃成为 Apache 软件基金会的顶级项目。
在德语中，Flink 一词表示快速和灵巧，项目采用一只松鼠的彩色图案作为 logo，这不仅是因为松鼠具有快速和灵巧的特点，还因为柏林的松鼠有一种迷人的红棕色，而 Flink 的松鼠 logo 拥有可爱的尾巴，尾巴的颜色与 Apache 软件基金会的 logo 颜色相呼应，也就是说，这是一只 Apache 风格的松鼠。 图 Flink Logo
Flink 主页在其顶部展示了该项目的理念：“Apache Flink 是为分布式、高性能、随时可用以及准确的流处理应用程序打造的开源流处理框架”。
Apache Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink 被设计在所有常见的集群环境中运行，以内存执行速度和任意规模来执行计算。
3. Flink 流处理特性
-
支持高吞吐、低延迟、高性能的流处理
-
支持带有事件时间的窗口（Window）操作
-
支持有状态计算的 Exactly-once 语义
-
支持高度灵活的窗口（Window）操作，支持基于 time、count、session，以及 data-driven 的窗口操作
-
支持具有 Backpressure 功能的持续流模型
-
支持基于轻量级分布式快照（Snapshot）实现的容错
-
一个运行时同时支持 Batch on Streaming 处理和 Streaming 处理
-
Flink 在 JVM 内部实现了自己的内存管理
-
支持迭代计算
-
支持程序自动优化：避免特定情况下 Shuffle、排序等昂贵操作，中间结果有必要进行缓存
4. Flink 基石
Flink 之所以能这么流行，离不开它最重要的四个基石：Checkpoint、State、Time、Window。
首先是 Checkpoint 机制，这是 Flink 最重要的一个特性。Flink 基于Chandy-Lamport算法实现了一个分布式的一致性的快照，从而提供了一致性的语义。Chandy-Lamport 算法实际上在 1985 年的时候已经被提出来，但并没有被很广泛的应用，而 Flink 则把这个算法发扬光大了。
Spark 最近在实现 Continue streaming，Continue streaming 的目的是为了降低它处理的延时，其也需要提供这种一致性的语义，最终采用 Chandy-Lamport 这个算法，说明 Chandy-Lamport 算法在业界得到了一定的肯定。
提供了一致性的语义之后，Flink 为了让用户在编程时能够更轻松、更容易地去管理状态，还提供了一套非常简单明了的 State API，包括里面的有 ValueState、ListState、MapState，近期添加了 BroadcastState，使用 State API 能够自动享受到这种一致性的语义。
除此之外，Flink 还实现了 Watermark 的机制，能够支持基于事件的时间的处理，或者说基于系统时间的处理，能够容忍数据的延时、容忍数据的迟到、容忍乱序的数据。
另外流计算中一般在对流数据进行操作之前都会先进行开窗，即基于一个什么样的窗口上做这个计算。Flink 提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义的窗口。
5. 批处理与流处理
批处理的特点是有界、持久、大量，批处理非常适合需要访问全套记录才能完成的计算工作，一般用于离线统计。流处理的特点是无界、实时，流处理方式无需针对整个数据集执行操作，而是对通过系统传输的每个数据项执行操作，一般用于实时统计。
在 Spark 生态体系中，对于批处理和流处理采用了不同的技术框架，批处理由 SparkSQL 实现，流处理由 Spark Streaming 实现，这也是大部分框架采用的策略，使用独立的处理器实现批处理和流处理，而 Flink 可以同时实现批处理和流处理。
Flink 是如何同时实现批处理与流处理的呢？答案是，Flink 将批处理（即处理有限的静态数据）视作一种特殊的流处理。
Flink 的核心计算架构是下图中的 Flink Runtime 执行引擎，它是一个分布式系统，能够接受数据流程序并在一台或多台机器上以容错方式执行。
Flink Runtime 执行引擎可以作为 YARN（Yet Another Resource Negotiator）的应用程序在集群上运行，也可以在 Mesos 集群上运行，还可以在单机上运行（这对于调试 Flink 应用程序来说非常有用）。
上图为 Flink 技术栈的核心组成部分，值得一提的是，Flink 分别提供了面向流式处理的接口（DataStream API）和面向批处理的接口（DataSet API）。因此，Flink 既可以完成流处理，也可以完成批处理。Flink 支持的拓展库涉及机器学习（FlinkML）、复杂事件处理（CEP）、以及图计算（Gelly），还有分别针对流处理和批处理的 Table API。
能被 Flink Runtime 执行引擎接受的程序很强大，但是这样的程序有着冗长的代码，编写起来也很费力，基于这个原因，Flink 提供了封装在 Runtime 执行引擎之上的 API，以帮助用户方便地生成流式计算程序。Flink 提供了用于流处理的 DataStream API 和用于批处理的 DataSet API。值得注意的是，尽管 Flink Runtime 执行引擎是基于流处理的，但是 DataSet API 先于 DataStream API 被开发出来，这是因为工业界对无限流处理的需求在 Flink 诞生之初并不大。
DataStream API 可以流畅地分析无限数据流，并且可以用 Java 或者 Scala 等来实现。开发人员需要基于一个叫 DataStream 的数据结构来开发，这个数据结构用于表示永不停止的分布式数据流。
Flink 的分布式特点体现在它能够在成百上千台机器上运行，它将大型的计算任务分成许多小的部分，每个机器执行一部分。Flink 能够自动地确保发生机器故障或者其他错误时计算能够持续进行，或者在修复 bug 或进行版本升级后有计划地再执行一次。这种能力使得开发人员不需要担心运行失败。Flink 本质上使用容错性数据流，这使得开发人员可以分析持续生成且永远不结束的数据（即流处理）。
二、Flink 部署及启动
Flink 支持多种安装模式:
-
local（本地）——单机模式，一般不使用；
-
standalone——独立模式，Flink 自带集群，开发测试环境使用；
-
yarn——计算资源统一由 Hadoop YARN 管理，生产环境使用。
Flink 集群的安装不属于本文档的范畴，如安装 Flink，可自行搜索资料进行安装。
本节重点在 Flink 的 Yarn 部署模式。
在一个企业中，为了最大化的利用集群资源，一般都会在一个集群中同时运行多种类型的 Workload，可以使用 YARN 来管理所有计算资源。
1. Flink 在 Yarn 上的部署架构
从图中可以看出，Yarn 的客户端需要获取 hadoop 的配置信息，连接 Yarn 的 ResourceManager。所以要设置 YARN_CONF_DIR 或者 HADOOP_CONF_DIR 或者 HADOOP_CONF_PATH，只要设置了其中一个环境变量，就会被读取。如果读取上述的变量失败了，那么将会选择 hadoop_home 的环境变量，会尝试加载$HADOOP_HOME/etc/hadoop 的配置文件。
-
当启动一个 Flink Yarn 会话时，客户端首先会检查本次请求的资源（存储、计算）是否足够。资源足够将会上传包含 HDFS 及 Flink 的配置信息和 Flink 的 jar 包到 HDFS；
-
客户端向 RM 发起请求；
-
RM 向 NM 发请求指令，创建 container，并从 HDFS 中下载 jar 以及配置文件；
-
启动 ApplicationMaster 和 jobmanager,将 jobmanager 的地址信息写到配置文件中，再发到 hdfs 上；
-
同时，AM 向 RM 发送心跳注册自己，申请资源（cpu、内存）；
-
创建 TaskManager 容器，从 HDFS 中下载 jar 包及配置文件并启动；
-
各 task 任务通过 jobmanager 汇报自己的状态和进度，AM 和 jobmanager 在一个容器上，AM 就能掌握各任务的运行状态，从而可以在任务失败时，重新启动任务；
-
任务完成后，AM 向 RM 注销并关闭自己；
2. 启动集群
- 修改 hadoop 的配置参数：vim etc/hadoop/yarn-site.xml
添加：
<property> <name>yarn.nodemanager.vmem-check-enabled</name> <value>false</value> </property>
修改 Hadoop 的 yarn-site.xml，添加该配置表示内存超过分配值，是否将任务杀掉。
默认为 true。 运行 Flink 程序，很容易内存超标，这个时候 yarn 会自动杀掉 job。
- 修改全局变量/etc/profile：
添加：export HADOOP_CONF_DIR=/export/servers/hadoop/etc/Hadoop
YARN_CONF_DIR 或者 HADOOP_CONF_DIR 必须将环境变量设置为读取 YARN 和 HDFS 配置
-
启动 HDFS、zookeeper（如果是外置 zookeeper）、YARN 集群；
-
使用 yarn-session 的模式提交作业。
Yarn Session 模式提交作业有两种方式：yarn-session 和 yarn-cluster
3. 模式一: yarn-session
特点：
-
使用 Flink 中的 yarn-session（yarn 客户端），会启动两个必要服务 JobManager 和 TaskManagers；
-
客户端通过 yarn-session 提交作业；
-
yarn-session 会一直启动，不停地接收客户端提交的任务；
-
如果拥有有大量的小作业，适合使用这种方式。
在 flink 目录启动 yarn-session：
bin/yarn-session.sh -n 2 -tm 800 -jm 800 -s 1 -d
-n 表示申请 2 个容器
-s 表示每个容器启动多少个 slot 离模式，表示以后台程
-tm 表示每个 TaskManager 申请 800M 内存
-d 分序方式运行
使用 flink 提交任务：
bin/flink run examples/batch/WordCount.jar
如果程序运行完了，可以使用 yarn application -kill application_id 杀掉任务：
yarn application -kill application_1554377097889_0002
bin/yarn-session.sh -n 2 -tm 800 -s 1 -d 意思是:
同时向 Yarn 申请 3 个 container（即便只申请了两个，因为 ApplicationMaster 和 Job Manager 有一个额外的容器。一旦将 Flink 部署到 YARN 群集中，它就会显示 Job Manager 的连接详细信息），其中 2 个 Container 启动 TaskManager（-n 2），每个 TaskManager 拥有两个 Task Slot（-s 1），并且向每个 TaskManager 的 Container 申请 800M 的内存，以及一个 ApplicationMaster（Job Manager）。
4. 模式二: yarn-cluster
特点：
-
直接提交任务给 YARN；
-
大作业，适合使用这种方式；
-
会自动关闭 session。
使用 flink 直接提交任务：
bin/flink run -m yarn-cluster -yn 2 -yjm 800 -ytm 800 /export/servers/flink-1.6.0/examples/batch/WordCount.jar
-yn 表示 TaskManager 的个数
注意：
- 在创建集群的时候，集群的配置参数就写好了，但是往往因为业务需要，要更改一些配置参数，这个时候可以不必因为一个实例的提交而修改 conf/flink-conf.yaml;
可以通过：-D <arg> Dynamic properties来覆盖原有的配置信息：比如：
-Dfs.overwrite-files=true -Dtaskmanager.network.numberOfBuffers=16368
- 如果使用的是 flink on yarn 方式，想切换回 standalone 模式的话，需要删除：/tmp/.yarn-properties-root，因为默认查找当前 yarn 集群中已有的 yarn-session 信息中的 jobmanager。
三、Flink 运行架构
1. Flink 程序结构
Flink 程序的基本构建块是流和转换（请注意，Flink 的 DataSet API 中使用的 DataSet 也是内部流 ）。从概念上讲，流是（可能永无止境的）数据记录流，而转换是将一个或多个流作为一个或多个流的操作。输入，并产生一个或多个输出流。
Flink 应用程序结构就是如上图所示：
Source: 数据源，Flink 在流处理和批处理上的 source 大概有 4 类：基于本地集合的 source、基于文件的 source、基于网络套接字的 source、自定义的 source。自定义的 source 常见的有 Apache kafka、RabbitMQ 等，当然你也可以定义自己的 source。
Transformation：数据转换的各种操作，有Map / FlatMap / Filter / KeyBy / Reduce / Fold / Aggregations / Window / WindowAll / Union / Window join / Split / Select等，操作很多，可以将数据转换计算成你想要的数据。
Sink：接收器，Flink 将转换计算后的数据发送的地点 ，你可能需要存储下来，Flink 常见的 Sink 大概有如下几类：写入文件、打印出来、写入 socket 、自定义的 sink 。自定义的 sink 常见的有 Apache kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem 等，同理你也可以定义自己的 sink。
2. Flink 并行数据流
Flink 程序在执行的时候，会被映射成一个 Streaming Dataflow，一个 Streaming Dataflow 是由一组 Stream 和 Transformation Operator 组成的。在启动时从一个或多个 Source Operator 开始，结束于一个或多个 Sink Operator。
Flink 程序本质上是并行的和分布式的，在执行过程中，一个流(stream)包含一个或多个流分区，而每一个 operator 包含一个或多个 operator 子任务。操作子任务间彼此独立，在不同的线程中执行，甚至是在不同的机器或不同的容器上。operator 子任务的数量是这一特定 operator 的并行度。相同程序中的不同 operator 有不同级别的并行度。
一个 Stream 可以被分成多个 Stream 的分区，也就是 Stream Partition。一个 Operator 也可以被分为多个 Operator Subtask。如上图中，Source 被分成 Source1 和 Source2，它们分别为 Source 的 Operator Subtask。每一个 Operator Subtask 都是在不同的线程当中独立执行的。一个 Operator 的并行度，就等于 Operator Subtask 的个数。上图 Source 的并行度为 2。而一个 Stream 的并行度就等于它生成的 Operator 的并行度。
数据在两个 operator 之间传递的时候有两种模式：
One to One 模式：两个 operator 用此模式传递的时候，会保持数据的分区数和数据的排序；如上图中的 Source1 到 Map1，它就保留的 Source 的分区特性，以及分区元素处理的有序性。
Redistributing （重新分配）模式：这种模式会改变数据的分区数；每个一个 operator subtask 会根据选择 transformation 把数据发送到不同的目标 subtasks,比如 keyBy()会通过 hashcode 重新分区,broadcast()和 rebalance()方***随机重新分区；
3. Task 和 Operator chain
Flink 的所有操作都称之为 Operator，客户端在提交任务的时候会对 Operator 进行优化操作，能进行合并的 Operator 会被合并为一个 Operator，合并后的 Operator 称为 Operator chain，实际上就是一个执行链，每个执行链会在 TaskManager 上一个独立的线程中执行。
4. 任务调度与执行
-
当 Flink 执行 executor 会自动根据程序代码生成 DAG 数据流图；
-
ActorSystem 创建 Actor 将数据流图发送给 JobManager 中的 Actor；
-
JobManager 会不断接收 TaskManager 的心跳消息，从而可以获取到有效的 TaskManager；
-
JobManager 通过调度器在 TaskManager 中调度执行 Task（在 Flink 中，最小的调度单元就是 task，对应就是一个线程）；
-
在程序运行过程中，task 与 task 之间是可以进行数据传输的。
Job Client：
- 主要职责是提交任务, 提交后可以结束进程, 也可以等待结果返回；
- Job Client 不是 Flink 程序执行的内部部分，但它是任务执行的起点；
- Job Client 负责接受用户的程序代码，然后创建数据流，将数据流提交给 Job Manager 以便进一步执行。 执行完成后，Job Client 将结果返回给用户。
JobManager：
- 主要职责是调度工作并协调任务做检查点；
- 集群中至少要有一个 master，master 负责调度 task，协调 checkpoints 和容错；
- 高可用设置的话可以有多个 master，但要保证一个是 leader, 其他是 standby；
- Job Manager 包含 Actor System、Scheduler、CheckPoint 三个重要的组件；
- JobManager 从客户端接收到任务以后, 首先生成优化过的执行计划, 再调度到 TaskManager 中执行。
TaskManager：
- 主要职责是从 JobManager 处接收任务, 并部署和启动任务, 接收上游的数据并处理；
- Task Manager 是在 JVM 中的一个或多个线程中执行任务的工作节点；
- TaskManager 在创建之初就设置好了 Slot, 每个 Slot 可以执行一个任务。
5. 任务槽和槽共享
每个 TaskManager 是一个 JVM 的进程, 可以在不同的线程中执行一个或多个子任务。 为了控制一个 worker 能接收多少个 task。worker 通过 task slot 来进行控制（一个 worker 至少有一个 task slot）。
1) 任务槽
每个 task slot 表示 TaskManager 拥有资源的一个固定大小的子集。
flink 将进程的内存进行了划分到多个 slot 中。
图中有 2 个 TaskManager，每个 TaskManager 有 3 个 slot 的，每个 slot 占有 1/3 的内存。
内存被划分到不同的 slot 之后可以获得如下好处:
-
TaskManager 最多能同时并发执行的任务是可以控制的，那就是 3 个，因为不能超过 slot 的数量。
-
slot 有独占的内存空间，这样在一个 TaskManager 中可以运行多个不同的作业，作业之间不受影响。
2) 槽共享
默认情况下，Flink 允许子任务共享插槽，即使它们是不同任务的子任务，只要它们来自同一个作业。结果是一个槽可以保存作业的整个管道。允许插槽共享有两个主要好处：
-
只需计算 Job 中最高并行度（parallelism）的 task slot,只要这个满足，其他的 job 也都能满足。
-
资源分配更加公平，如果有比较空闲的 slot 可以将更多的任务分配给它。图中若没有任务槽共享，负载不高的 Source/Map 等 subtask 将会占据许多资源，而负载较高的窗口 subtask 则会缺乏资源。
-
有了任务槽共享，可以将基本并行度（base parallelism）从 2 提升到 6.提高了分槽资源的利用率。同时它还可以保障 TaskManager 给 subtask 的分配的 slot 方案更加公平。
四、Flink 算子大全
Flink 和 Spark 类似，也是一种一站式处理的框架；既可以进行批处理（DataSet），也可以进行实时处理（DataStream）。
所以下面将 Flink 的算子分为两大类：一类是 DataSet，一类是 DataStream。
DataSet 批处理算子
一、Source 算子
1. fromCollection
fromCollection：从本地集合读取数据
例：
val env = ExecutionEnvironment.getExecutionEnvironment val textDataSet: DataSet[String] = env.fromCollection( List("1,张三", "2,李四", "3,王五", "4,赵六") )
2. readTextFile
readTextFile：从文件中读取
val textDataSet: DataSet[String] = env.readTextFile("/data/a.txt")
3. readTextFile：遍历目录
readTextFile 可以对一个文件目录内的所有文件，包括所有子目录中的所有文件的遍历访问方式
val parameters = new Configuration // recursive.file.enumeration 开启递归 parameters.setBoolean("recursive.file.enumeration", true) val file = env.readTextFile("/data").withParameters(parameters)
4. readTextFile：读取压缩文件
对于以下压缩类型，不需要指定任何额外的 inputformat 方法，flink 可以自动识别并且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。
| 压缩方法 | 文件扩展名 | 是否可并行读取 |
|---|---|---|
| DEFLATE | .deflate | no |
| GZip | .gz .gzip | no |
| Bzip2 | .bz2 | no |
| XZ | .xz | no |
val file = env.readTextFile("/data/file.gz")
二、Transform 转换算子
因为 Transform 算子基于 Source 算子操作，所以首先构建 Flink 执行环境及 Source 算子，后续 Transform 算子操作基于此：
val env = ExecutionEnvironment.getExecutionEnvironment val textDataSet: DataSet[String] = env.fromCollection( List("张三,1", "李四,2", "王五,3", "张三,4") )
1. map
将 DataSet 中的每一个元素转换为另外一个元素
// 使用map将List转换为一个Scala的样例类 case class User(name: String, id: String) val userDataSet: DataSet[User] = textDataSet.map { text => val fieldArr = text.split(",") User(fieldArr(0), fieldArr(1)) } userDataSet.print()
2. flatMap
将 DataSet 中的每一个元素转换为 0...n 个元素。
// 使用flatMap操作，将集合中的数据： // 根据第一个元素，进行分组 // 根据第二个元素，进行聚合求值 val result = textDataSet.flatMap(line => line) .groupBy(0) // 根据第一个元素，进行分组 .sum(1) // 根据第二个元素，进行聚合求值 result.print()
3. mapPartition
将一个分区中的元素转换为另一个元素
// 使用mapPartition操作，将List转换为一个scala的样例类 case class User(name: String, id: String) val result: DataSet[User] = textDataSet.mapPartition(line => { line.map(index => User(index._1, index._2)) }) result.print()
4. filter
过滤出来一些符合条件的元素，返回boolean 值为 true的元素
val source: DataSet[String] = env.fromElements("java", "scala", "java") val filter:DataSet[String] = source.filter(line => line.contains("java"))//过滤出带java的数据 filter.print()
5. reduce
可以对一个 dataset 或者一个 group 来进行聚合计算，最终聚合成一个元素
// 使用 fromElements 构建数据源 val source = env.fromElements(("java", 1), ("scala", 1), ("java", 1)) // 使用map转换成DataSet元组 val mapData: DataSet[(String, Int)] = source.map(line => line) // 根据首个元素分组 val groupData = mapData.groupBy(_._1) // 使用reduce聚合 val reduceData = groupData.reduce((x, y) => (x._1, x._2 + y._2)) // 打印测试 reduceData.print()
6. reduceGroup
将一个 dataset 或者一个 group聚合成一个或多个元素。
reduceGroup 是 reduce 的一种优化方案；
它会先分组 reduce，然后在做整体的 reduce；这样做的好处就是可以减少网络 IO
// 使用 fromElements 构建数据源 val source: DataSet[(String, Int)] = env.fromElements(("java", 1), ("scala", 1), ("java", 1)) // 根据首个元素分组 val groupData = source.groupBy(_._1) // 使用reduceGroup聚合 val result: DataSet[(String, Int)] = groupData.reduceGroup { (in: Iterator[(String, Int)], out: Collector[(String, Int)]) => val tuple = in.reduce((x, y) => (x._1, x._2 + y._2)) out.collect(tuple) } // 打印测试 result.print()
7. minBy 和 maxBy
选择具有最小值或最大值的元素
// 使用minBy操作，求List中每个人的最小值 // List("张三,1", "李四,2", "王五,3", "张三,4") case class User(name: String, id: String) // 将List转换为一个scala的样例类 val text: DataSet[User] = textDataSet.mapPartition(line => { line.map(index => User(index._1, index._2)) }) val result = text .groupBy(0) // 按照姓名分组 .minBy(1) // 每个人的最小值
8. Aggregate
在数据集上进行聚合求最值（最大值、最小值）
val data = new mutable.MutableList[(Int, String, Double)] data.+=((1, "yuwen", 89.0)) data.+=((2, "shuxue", 92.2)) data.+=((3, "yuwen", 89.99)) // 使用 fromElements 构建数据源 val input: DataSet[(Int, String, Double)] = env.fromCollection(data) // 使用group执行分组操作 val value = input.groupBy(1) // 使用aggregate求最大值元素 .aggregate(Aggregations.MAX, 2) // 打印测试 value.print()
Aggregate 只能作用于元组上
注意：
要使用 aggregate，只能使用字段索引名或索引名称来进行分组groupBy(0)，否则会报一下错误:
Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.
9. distinct
去除重复的数据
// 数据源使用上一题的 // 使用distinct操作，根据科目去除集合中重复的元组数据 val value: DataSet[(Int, String, Double)] = input.distinct(1) value.print()
10. first
取前 N 个数
input.first(2) // 取前两个数
11. join
将两个 DataSet 按照一定条件连接到一起，形成新的 DataSet
// s1 和 s2 数据集格式如下： // DataSet[(Int, String,String, Double)] val joinData = s1.join(s2) // s1数据集 join s2数据集 .where(0).equalTo(0) { // join的条件 (s1, s2) => (s1._1, s1._2, s2._2, s1._3) }
12. leftOuterJoin
左外连接,左边的 Dataset 中的每一个元素，去连接右边的元素
此外还有：
rightOuterJoin：右外连接,左边的 Dataset 中的每一个元素，去连接左边的元素
fullOuterJoin：全外连接,左右两边的元素，全部连接
下面以 leftOuterJoin 进行示例：
val data1 = ListBuffer[Tuple2[Int,String]]() data1.append((1,"zhangsan")) data1.append((2,"lisi")) data1.append((3,"wangwu")) data1.append((4,"zhaoliu")) val data2 = ListBuffer[Tuple2[Int,String]]() data2.append((1,"beijing")) data2.append((2,"shanghai")) data2.append((4,"guangzhou")) val text1 = env.fromCollection(data1) val text2 = env.fromCollection(data2) text1.leftOuterJoin(text2).where(0).equalTo(0).apply((first,second)=>{ if(second==null){ (first._1,first._2,"null") }else{ (first._1,first._2,second._2) } }).print()
13. cross
交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集
和 join 类似，但是这种交叉操作会产生笛卡尔积，在数据比较大的时候，是非常消耗内存的操作
val cross = input1.cross(input2){
(input1 , input2) => (input1._1,input1._2,input1._3,input2._2)
}
cross.print() 14. union
联合操作，创建包含来自该数据集和其他数据集的元素的新数据集,不会去重
val unionData: DataSet[String] = elements1.union(elements2).union(elements3) // 去除重复数据 val value = unionData.distinct(line => line)
15. rebalance
Flink 也有数据倾斜的时候，比如当前有数据量大概 10 亿条数据需要处理，在处理过程中可能会发生如图所示的状况：
这个时候本来总体数据量只需要 10 分钟解决的问题，出现了数据倾斜，机器 1 上的任务需要 4 个小时才能完成，那么其他 3 台机器执行完毕也要等待机器 1 执行完毕后才算整体将任务完成； 所以在实际的工作中，出现这种情况比较好的解决方案就是接下来要介绍的—rebalance（内部使用 round robin 方法将数据均匀打散。这对于数据倾斜时是很好的选择。）
// 使用rebalance操作，避免数据倾斜 val rebalance = filterData.rebalance()
16. partitionByHash
按照指定的 key 进行 hash 分区
val data = new mutable.MutableList[(Int, Long, String)] data.+=((1, 1L, "Hi")) data.+=((2, 2L, "Hello")) data.+=((3, 2L, "Hello world")) val collection = env.fromCollection(data) val unique = collection.partitionByHash(1).mapPartition{ line => line.map(x => (x._1 , x._2 , x._3)) } unique.writeAsText("hashPartition", WriteMode.NO_OVERWRITE) env.execute()
17. partitionByRange
根据指定的 key 对数据集进行范围分区
val data = new mutable.MutableList[(Int, Long, String)] data.+=((1, 1L, "Hi")) data.+=((2, 2L, "Hello")) data.+=((3, 2L, "Hello world")) data.+=((4, 3L, "Hello world, how are you?")) val collection = env.fromCollection(data) val unique = collection.partitionByRange(x => x._1).mapPartition(line => line.map{ x=> (x._1 , x._2 , x._3) }) unique.writeAsText("rangePartition", WriteMode.OVERWRITE) env.execute()
18. sortPartition
根据指定的字段值进行分区的排序
val data = new mutable.MutableList[(Int, Long, String)] data.+=((1, 1L, "Hi")) data.+=((2, 2L, "Hello")) data.+=((3, 2L, "Hello world")) data.+=((4, 3L, "Hello world, how are you?")) val ds = env.fromCollection(data) val result = ds .map { x => x }.setParallelism(2) .sortPartition(1, Order.DESCENDING)//第一个参数代表按照哪个字段进行分区 .mapPartition(line => line) .collect() println(result)
三、Sink 算子
1. collect
将数据输出到本地集合
result.collect()
2. writeAsText
将数据输出到文件
Flink 支持多种存储设备上的文件，包括本地文件，hdfs 文件等
Flink 支持多种文件的存储格式，包括 text 文件，CSV 文件等
// 将数据写入本地文件 result.writeAsText("/data/a", WriteMode.OVERWRITE) // 将数据写入HDFS result.writeAsText("hdfs://node01:9000/data/a", WriteMode.OVERWRITE)
DataStream 流处理算子
和 DataSet 一样，DataStream 也包括一系列的 Transformation 操作
一、Source 算子
Flink 可以使用 StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源。
Flink 已经提供了若干实现好了的 source functions，当然我们也可以通过实现 SourceFunction 来自定义非并行的 source 或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source。
Flink 在流处理上的 source 和在批处理上的 source 基本一致。大致有 4 大类：
- 基于本地集合的 source（Collection-based-source）
- 基于文件的 source（File-based-source）- 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回
- 基于网络套接字的 source（Socket-based-source）- 从 socket 读取。元素可以用分隔符切分。
- 自定义的 source（Custom-source）
下面使用 addSource 将 Kafka 数据写入 Flink 为例：
如果需要外部数据源对接，可使用 addSource，如将 Kafka 数据写入 Flink， 先引入依赖：
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11 --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka-0.11_2.11</artifactId> <version>1.10.0</version> </dependency>
将 Kafka 数据写入 Flink：
val properties = new Properties() properties.setProperty("bootstrap.servers", "localhost:9092") properties.setProperty("group.id", "consumer-group") properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer") properties.setProperty("auto.offset.reset", "latest") val source = env.addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
基于网络套接字的：
val source = env.socketTextStream("IP", PORT)
二、Transform 转换算子
1. map
将 DataSet 中的每一个元素转换为另外一个元素
dataStream.map { x => x * 2 } 2. FlatMap
采用一个数据元并生成零个，一个或多个数据元。将句子分割为单词的 flatmap 函数
dataStream.flatMap { str => str.split(" ") } 3. Filter
计算每个数据元的布尔函数，并保存函数返回 true 的数据元。过滤掉零值的过滤器
dataStream.filter { _ != 0 } 4. KeyBy
逻辑上将流分区为不相交的分区。具有相同 Keys 的所有记录都分配给同一分区。在内部，keyBy（）是使用散列分区实现的。指定键有不同的方法。
此转换返回 KeyedStream，其中包括使用被 Keys 化状态所需的 KeyedStream。
dataStream.keyBy(0) 5. Reduce
被 Keys 化数据流上的“滚动”Reduce。将当前数据元与最后一个 Reduce 的值组合并发出新值
keyedStream.reduce { _ + _ } 6. Fold
具有初始值的被 Keys 化数据流上的“滚动”折叠。将当前数据元与最后折叠的值组合并发出新值
val result: DataStream[String] = keyedStream.fold("start")((str, i) => { str + "-" + i }) // 解释：当上述代码应用于序列（1,2,3,4,5）时，输出结果“start-1”，“start-1-2”，“start-1-2-3”，...
7. Aggregations
在被 Keys 化数据流上滚动聚合。min 和 minBy 之间的差异是 min 返回最小值，而 minBy 返回该字段中具有最小值的数据元（max 和 maxBy 相同）。
keyedStream.sum(0); keyedStream.min(0); keyedStream.max(0); keyedStream.minBy(0); keyedStream.maxBy(0);
8. Window
可以在已经分区的 KeyedStream 上定义 Windows。Windows 根据某些特征（例如，在最后 5 秒内到达的数据）对每个 Keys 中的数据进行分组。这里不再对窗口进行详解，有关窗口的完整说明，请查看这篇文章： Flink 中极其重要的 Time 与 Window 详细解析
dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5)));
9. WindowAll
Windows 可以在常规 DataStream 上定义。Windows 根据某些特征（例如，在最后 5 秒内到达的数据）对所有流事件进行分组。
注意：在许多情况下，这是非并行转换。所有记录将收集在 windowAll 算子的一个任务中。
dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
10. Window Apply
将一般函数应用于整个窗口。
注意：如果您正在使用 windowAll 转换，则需要使用 AllWindowFunction。
下面是一个手动求和窗口数据元的函数
windowedStream.apply { WindowFunction }
allWindowedStream.apply { AllWindowFunction } 11. Window Reduce
将函数缩减函数应用于窗口并返回缩小的值
windowedStream.reduce { _ + _ } 12. Window Fold
将函数折叠函数应用于窗口并返回折叠值
val result: DataStream[String] = windowedStream.fold("start", (str, i) => { str + "-" + i }) // 上述代码应用于序列（1,2,3,4,5）时，将序列折叠为字符串“start-1-2-3-4-5”
13. Union
两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次数据元
dataStream.union(otherStream1, otherStream2, ...)
14. Window Join
在给定 Keys 和公共窗口上连接两个数据流
dataStream.join(otherStream) .where(<key selector>).equalTo(<key selector>) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new JoinFunction () {...})
15. Interval Join
在给定的时间间隔内使用公共 Keys 关联两个被 Key 化的数据流的两个数据元 e1 和 e2，以便 e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound
am.intervalJoin(otherKeyedStream) .between(Time.milliseconds(-2), Time.milliseconds(2)) .upperBoundExclusive(true) .lowerBoundExclusive(true) .process(new IntervalJoinFunction() {...})
16. Window CoGroup
在给定 Keys 和公共窗口上对两个数据流进行 Cogroup
dataStream.coGroup(otherStream) .where(0).equalTo(1) .window(TumblingEventTimeWindows.of(Time.seconds(3))) .apply (new CoGroupFunction () {...})
17. Connect
“连接”两个保存其类型的数据流。连接允许两个流之间的共享状态
DataStream<Integer> someStream = ... DataStream<String> otherStream = ... ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream) // ... 代表省略中间操作
18. CoMap，CoFlatMap
类似于连接数据流上的 map 和 flatMap
connectedStreams.map( (_ : Int) => true, (_ : String) => false)connectedStreams.flatMap( (_ : Int) => true, (_ : String) => false)
19. Split
根据某些标准将流拆分为两个或更多个流
val split = someDataStream.split( (num: Int) => (num % 2) match { case 0 => List("even") case 1 => List("odd") })
20. Select
从拆分流中选择一个或多个流
SplitStream<Integer> split;DataStream<Integer> even = split.select("even");DataStream<Integer> odd = split.select("odd");DataStream<Integer> all = split.select("even","odd")
三、Sink 算子
支持将数据输出到：
- 本地文件(参考批处理)
- 本地集合(参考批处理)
- HDFS(参考批处理)
除此之外，还支持：
- sink 到 kafka
- sink 到 mysql
- sink 到 redis
下面以 sink 到 kafka 为例：
val sinkTopic = "test" //样例类 case class Student(id: Int, name: String, addr: String, sex: String) val mapper: ObjectMapper = new ObjectMapper() //将对象转换成字符串 def toJsonString(T: Object): String = { mapper.registerModule(DefaultScalaModule) mapper.writeValueAsString(T) } def main(args: Array[String]): Unit = { //1.创建流执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.准备数据 val dataStream: DataStream[Student] = env.fromElements( Student(8, "xiaoming", "beijing biejing", "female") ) //将student转换成字符串 val studentStream: DataStream[String] = dataStream.map(student => toJsonString(student) // 这里需要显示SerializerFeature中的某一个，否则会报同时匹配两个方法的错误 ) //studentStream.print() val prop = new Properties() prop.setProperty("bootstrap.servers", "node01:9092") val myProducer = new FlinkKafkaProducer011[String](sinkTopic, new KeyedSerializationSchemaWrapper[String](new SimpleStringSchema()), prop) studentStream.addSink(myProducer) studentStream.print() env.execute("Flink add sink") }
本文档首发于公众号【五分钟学大数据】，更多大数据技术文档可下方扫码关注获取：
五、流处理中的 Time 与 Window
Flink 是流式的、实时的 计算引擎。
上面一句话就有两个概念，一个是流式，一个是实时。
流式：就是数据源源不断的流进来，也就是数据没有边界，但是我们计算的时候必须在一个有边界的范围内进行，所以这里面就有一个问题，边界怎么确定？ 无非就两种方式，根据时间段或者数据量进行确定，根据时间段就是每隔多长时间就划分一个边界，根据数据量就是每来多少条数据划分一个边界，Flink 中就是这么划分边界的，本文会详细讲解。
实时：就是数据发送过来之后立马就进行相关的计算，然后将结果输出。这里的计算有两种：
-
一种是只有边界内的数据进行计算，这种好理解，比如统计每个用户最近五分钟内浏览的新闻数量，就可以取最近五分钟内的所有数据，然后根据每个用户分组，统计新闻的总数。
-
另一种是边界内数据与外部数据进行关联计算，比如：统计最近五分钟内浏览新闻的用户都是来自哪些地区，这种就需要将五分钟内浏览新闻的用户信息与 hive 中的地区维表进行关联，然后在进行相关计算。
本节所讲的 Flink 内容就是围绕以上概念进行详细剖析的！
1. Time
在 Flink 中，如果以时间段划分边界的话，那么时间就是一个极其重要的字段。
Flink 中的时间有三种类型，如下图所示：
-
Event Time：是事件创建的时间。它通常由事件中的时间戳描述，例如采集的日志数据中，每一条日志都会记录自己的生成时间，Flink 通过时间戳分配器访问事件时间戳。
-
Ingestion Time：是数据进入 Flink 的时间。
-
Processing Time：是每一个执行基于时间操作的算子的本地系统时间，与机器相关，默认的时间属性就是 Processing Time。
例如，一条日志进入 Flink 的时间为 2021-01-22 10:00:00.123，到达 Window 的系统时间为 2021-01-22 10:00:01.234，日志的内容如下：
2021-01-06 18:37:15.624 INFO Fail over to rm2
对于业务来说，要统计 1min 内的故障日志个数，哪个时间是最有意义的？—— eventTime，因为我们要根据日志的生成时间进行统计。
2. Window
Window，即窗口，我们前面一直提到的边界就是这里的 Window(窗口)。
官方解释：流式计算是一种被设计用于处理无限数据集的数据处理引擎，而无限数据集是指一种不断增长的本质上无限的数据集，而 window 是一种切割无限数据为有限块进行处理的手段。
所以Window 是无限数据流处理的核心，Window 将一个无限的 stream 拆分成有限大小的”buckets”桶，我们可以在这些桶上做计算操作。
Window 类型
本文刚开始提到，划分窗口就两种方式：
- 根据时间进行截取(time-driven-window)，比如每 1 分钟统计一次或每 10 分钟统计一次。
- 根据数据进行截取(data-driven-window)，比如每 5 个数据统计一次或每 50 个数据统计一次。
窗口类型 对于 TimeWindow(根据时间划分窗口)， 可以根据窗口实现原理的不同分成三类：滚动窗口（Tumbling Window）、滑动窗口（Sliding Window）和会话窗口（Session Window）。
- 滚动窗口（Tumbling Windows）
将数据依据固定的窗口长度对数据进行切片。
特点：时间对齐，窗口长度固定，没有重叠。
滚动窗口分配器将每个元素分配到一个指定窗口大小的窗口中，滚动窗口有一个固定的大小，并且不会出现重叠。
例如：如果你指定了一个 5 分钟大小的滚动窗口，窗口的创建如下图所示：
滚动窗口 适用场景：适合做 BI 统计等（做每个时间段的聚合计算）。
- 滑动窗口（Sliding Windows）
滑动窗口是固定窗口的更广义的一种形式，滑动窗口由固定的窗口长度和滑动间隔组成。
特点：时间对齐，窗口长度固定，有重叠。
滑动窗口分配器将元素分配到固定长度的窗口中，与滚动窗口类似，窗口的大小由窗口大小参数来配置，另一个窗口滑动参数控制滑动窗口开始的频率。因此，滑动窗口如果滑动参数小于窗口大小的话，窗口是可以重叠的，在这种情况下元素会被分配到多个窗口中。
例如，你有 10 分钟的窗口和 5 分钟的滑动，那么每个窗口中 5 分钟的窗口里包含着上个 10 分钟产生的数据，如下图所示：
滑动窗口 适用场景：对最近一个时间段内的统计（求某接口最近 5min 的失败率来决定是否要报警）。
- 会话窗口（Session Windows）
由一系列事件组合一个指定时间长度的 timeout 间隙组成，类似于 web 应用的 session，也就是一段时间没有接收到新数据就会生成新的窗口。
特点：时间无对齐。
session 窗口分配器通过 session 活动来对元素进行分组，session 窗口跟滚动窗口和滑动窗口相比，不会有重叠和固定的开始时间和结束时间的情况，相反，当它在一个固定的时间周期内不再收到元素，即非活动间隔产生，那个这个窗口就会关闭。一个 session 窗口通过一个 session 间隔来配置，这个 session 间隔定义了非活跃周期的长度，当这个非活跃周期产生，那么当前的 session 将关闭并且后续的元素将被分配到新的 session 窗口中去。
会话窗口 3. Window API
1) TimeWindow
TimeWindow 是将指定时间范围内的所有数据组成一个 window，一次对一个 window 里面的所有数据进行计算（就是本文开头说的对一个边界内的数据进行计算）。
我们以 红绿灯路口通过的汽车数量 为例子：
红绿灯路口会有汽车通过，一共会有多少汽车通过，无法计算。因为车流源源不断，计算没有边界。
所以我们统计每 15 秒钟通过红路灯的汽车数量，如第一个 15 秒为 2 辆，第二个 15 秒为 3 辆，第三个 15 秒为 1 辆 ...
- tumbling-time-window (无重叠数据)
我们使用 Linux 中的 nc 命令模拟数据的发送方
1.开启发送端口，端口号为9999 nc -lk 9999 2.发送内容（key 代表不同的路口，value 代表每次通过的车辆） 一次发送一行，发送的时间间隔代表汽车经过的时间间隔 9,3 9,2 9,7 4,9 2,6 1,5 2,3 5,7 5,4
Flink 进行采集数据并计算：
object Window { def main(args: Array[String]): Unit = { //TODO time-window //1.创建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.定义数据流来源 val text = env.socketTextStream("localhost", 9999) //3.转换数据格式，text->CarWc case class CarWc(sensorId: Int, carCnt: Int) val ds1: DataStream[CarWc] = text.map { line => { val tokens = line.split(",") CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt) } } //4.执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5秒 //也就是说，每5秒钟统计一次，在这过去的5秒钟内，各个路口通过红绿灯汽车的数量。 val ds2: DataStream[CarWc] = ds1 .keyBy("sensorId") .timeWindow(Time.seconds(5)) .sum("carCnt") //5.显示统计结果 ds2.print() //6.触发流计算 env.execute(this.getClass.getName) } }
我们发送的数据并没有指定时间字段，所以 Flink 使用的是默认的 Processing Time，也就是 Flink 系统处理数据时的时间。
- sliding-time-window (有重叠数据)
//1.创建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.定义数据流来源 val text = env.socketTextStream("localhost", 9999) //3.转换数据格式，text->CarWc case class CarWc(sensorId: Int, carCnt: Int) val ds1: DataStream[CarWc] = text.map { line => { val tokens = line.split(",") CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt) } } //4.执行统计操作，每个sensorId一个sliding窗口，窗口时间10秒,滑动时间5秒 //也就是说，每5秒钟统计一次，在这过去的10秒钟内，各个路口通过红绿灯汽车的数量。 val ds2: DataStream[CarWc] = ds1 .keyBy("sensorId") .timeWindow(Time.seconds(10), Time.seconds(5)) .sum("carCnt") //5.显示统计结果 ds2.print() //6.触发流计算 env.execute(this.getClass.getName)
2) CountWindow
CountWindow 根据窗口中相同 key 元素的数量来触发执行，执行时只计算元素数量达到窗口大小的 key 对应的结果。
注意：CountWindow 的 window_size 指的是相同 Key 的元素的个数，不是输入的所有元素的总数。
- tumbling-count-window (无重叠数据)
//1.创建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.定义数据流来源 val text = env.socketTextStream("localhost", 9999) //3.转换数据格式，text->CarWc case class CarWc(sensorId: Int, carCnt: Int) val ds1: DataStream[CarWc] = text.map { (f) => { val tokens = f.split(",") CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt) } } //4.执行统计操作，每个sensorId一个tumbling窗口，窗口的大小为5 //按照key进行收集，对应的key出现的次数达到5次作为一个结果 val ds2: DataStream[CarWc] = ds1 .keyBy("sensorId") .countWindow(5) .sum("carCnt") //5.显示统计结果 ds2.print() //6.触发流计算 env.execute(this.getClass.getName)
- sliding-count-window (有重叠数据)
同样也是窗口长度和滑动窗口的操作：窗口长度是 5，滑动长度是 3
//1.创建运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2.定义数据流来源 val text = env.socketTextStream("localhost", 9999) //3.转换数据格式，text->CarWc case class CarWc(sensorId: Int, carCnt: Int) val ds1: DataStream[CarWc] = text.map { (f) => { val tokens = f.split(",") CarWc(tokens(0).trim.toInt, tokens(1).trim.toInt) } } //4.执行统计操作，每个sensorId一个sliding窗口，窗口大小3条数据,窗口滑动为3条数据 //也就是说，每个路口分别统计，收到关于它的3条消息时统计在最近5条消息中，各自路口通过的汽车数量 val ds2: DataStream[CarWc] = ds1 .keyBy("sensorId") .countWindow(5, 3) .sum("carCnt") //5.显示统计结果 ds2.print() //6.触发流计算 env.execute(this.getClass.getName)
- Window 总结
-
flink 支持两种划分窗口的方式（time 和 count）
-
如果根据时间划分窗口，那么它就是一个 time-window
-
如果根据数据划分窗口，那么它就是一个 count-window
-
-
flink 支持窗口的两个重要属性（size 和 interval）
-
如果 size=interval,那么就会形成 tumbling-window(无重叠数据)
-
如果 size>interval,那么就会形成 sliding-window(有重叠数据)
-
如果 size<interval,那么这种窗口将会丢失数据。比如每 5 秒钟，统计过去 3 秒的通过路口汽车的数据，将会漏掉 2 秒钟的数据。
-
-
通过组合可以得出四种基本窗口
-
time-tumbling-window 无重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5))
-
time-sliding-window 有重叠数据的时间窗口，设置方式举例：timeWindow(Time.seconds(5), Time.seconds(3))
-
count-tumbling-window 无重叠数据的数量窗口，设置方式举例：countWindow(5)
-
count-sliding-window 有重叠数据的数量窗口，设置方式举例：countWindow(5,3)
-
3) Window Reduce
WindowedStream → DataStream：给 window 赋一个 reduce 功能的函数，并返回一个聚合的结果。
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time object StreamWindowReduce { def main(args: Array[String]): Unit = { // 获取执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 创建SocketSource val stream = env.socketTextStream("node01", 9999) // 对stream进行处理并按key聚合 val streamKeyBy = stream.map(item => (item, 1)).keyBy(0) // 引入时间窗口 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5)) // 执行聚合操作 val streamReduce = streamWindow.reduce( (item1, item2) => (item1._1, item1._2 + item2._2) ) // 将聚合数据写入文件 streamReduce.print() // 执行程序 env.execute("TumblingWindow") } }
4) Window Apply
apply 方法可以进行一些自定义处理，通过匿名内部类的方法来实现。当有一些复杂计算时使用。
用法
- 实现一个 WindowFunction 类
- 指定该类的泛型为 [输入数据类型, 输出数据类型, keyBy 中使用分组字段的类型, 窗口类型]
示例：使用 apply 方法来实现单词统计
步骤：
- 获取流处理运行环境
- 构建 socket 流数据源，并指定 IP 地址和端口号
- 对接收到的数据转换成单词元组
- 使用 keyBy 进行分流（分组）
- 使用 timeWinodw 指定窗口的长度（每 3 秒计算一次）
- 实现一个 WindowFunction 匿名内部类
- apply 方法中实现聚合计算
- 使用 Collector.collect 收集数据
核心代码如下：
//1. 获取流处理运行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //2. 构建socket流数据源，并指定IP地址和端口号 val textDataStream = env.socketTextStream("node01", 9999).flatMap(_.split(" ")) //3. 对接收到的数据转换成单词元组 val wordDataStream = textDataStream.map(_->1) //4. 使用 keyBy 进行分流（分组） val groupedDataStream: KeyedStream[(String, Int), String] = wordDataStream.keyBy(_._1) //5. 使用 timeWinodw 指定窗口的长度（每3秒计算一次） val windowDataStream: WindowedStream[(String, Int), String, TimeWindow] = groupedDataStream.timeWindow(Time.seconds(3)) //6. 实现一个WindowFunction匿名内部类 val reduceDatStream: DataStream[(String, Int)] = windowDataStream.apply(new RichWindowFunction[(String, Int), (String, Int), String, TimeWindow] { //在apply方法中实现数据的聚合 override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = { println("hello world") val tuple = input.reduce((t1, t2) => { (t1._1, t1._2 + t2._2) }) //将要返回的数据收集起来，发送回去 out.collect(tuple) } }) reduceDatStream.print() env.execute()
5) Window Fold
WindowedStream → DataStream：给窗口赋一个 fold 功能的函数，并返回一个 fold 后的结果。
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.api.scala._ import org.apache.flink.streaming.api.windowing.time.Time object StreamWindowFold { def main(args: Array[String]): Unit = { // 获取执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 创建SocketSource val stream = env.socketTextStream("node01", 9999,'\n',3) // 对stream进行处理并按key聚合 val streamKeyBy = stream.map(item => (item, 1)).keyBy(0) // 引入滚动窗口 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5)) // 执行fold操作 val streamFold = streamWindow.fold(100){ (begin, item) => begin + item._2 } // 将聚合数据写入文件 streamFold.print() // 执行程序 env.execute("TumblingWindow") } }
6) Aggregation on Window
WindowedStream → DataStream：对一个 window 内的所有元素做聚合操作。min 和 minBy 的区别是 min 返回的是最小值，而 minBy 返回的是包含最小值字段的元素(同样的原理适用于 max 和 maxBy)。
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.windowing.time.Time import org.apache.flink.api.scala._ object StreamWindowAggregation { def main(args: Array[String]): Unit = { // 获取执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment // 创建SocketSource val stream = env.socketTextStream("node01", 9999) // 对stream进行处理并按key聚合 val streamKeyBy = stream.map(item => (item.split(" ")(0), item.split(" ")(1))).keyBy(0) // 引入滚动窗口 val streamWindow = streamKeyBy.timeWindow(Time.seconds(5)) // 执行聚合操作 val streamMax = streamWindow.max(1) // 将聚合数据写入文件 streamMax.print() // 执行程序 env.execute("TumblingWindow") } }
4. EventTime 与 Window
1) EventTime 的引入
- 与现实世界中的时间是不一致的，在 flink 中被划分为事件时间，提取时间，处理时间三种。
- 如果以 EventTime 为基准来定义时间窗口那将形成 EventTimeWindow,要求消息本身就应该携带 EventTime
- 如果以 IngesingtTime 为基准来定义时间窗口那将形成 IngestingTimeWindow,以 source 的 systemTime 为准。
- 如果以 ProcessingTime 基准来定义时间窗口那将形成 ProcessingTimeWindow，以 operator 的 systemTime 为准。
在 Flink 的流式处理中，绝大部分的业务都会使用 eventTime，一般只在 eventTime 无法使用时，才会被迫使用 ProcessingTime 或者 IngestionTime。
如果要使用 EventTime，那么需要引入 EventTime 的时间属性，引入方式如下所示：
val env = StreamExecutionEnvironment.getExecutionEnvironment // 从调用时刻开始给env创建的每一个stream追加时间特征 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
2) Watermark
我们知道，流处理从事件产生，到流经 source，再到 operator，中间是有一个过程和时间的，虽然大部分情况下，流到 operator 的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、背压等原因，导致乱序的产生，所谓乱序，就是指 Flink 接收到的事件的先后顺序不是严格按照事件的 Event Time 顺序排列的，所以 Flink 最初设计的时候，就考虑到了网络延迟，网络乱序等问题，所以提出了一个抽象概念：水印（WaterMark）；
如上图所示，就出现一个问题，一旦出现乱序，如果只根据 EventTime 决定 Window 的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发 Window 去进行计算了，这个特别的机制，就是 Watermark。
Watermark 是用于处理乱序事件的，而正确的处理乱序事件，通常用 Watermark 机制结合 Window 来实现。
数据流中的 Watermark 用于表示 timestamp 小于 Watermark 的数据，都已经到达了，因此，Window 的执行也是由 Watermark 触发的。
Watermark 可以理解成一个延迟触发机制，我们可以设置 Watermark 的延时时长 t，每次系统会校验已经到达的数据中最大的 maxEventTime，然后认定 EventTime 小于 maxEventTime - t 的所有数据都已经到达，如果有窗口的停止时间等于 maxEventTime – t，那么这个窗口被触发执行。
有序流的 Watermarker 如下图所示：（Watermark 设置为 0）
有序数据的Watermark 乱序流的 Watermarker 如下图所示：（Watermark 设置为 2）
无序数据的Watermark 当 Flink 接收到每一条数据时，都会产生一条 Watermark，这条 Watermark 就等于当前所有到达数据中的 maxEventTime - 延迟时长，也就是说，Watermark 是由数据携带的，一旦数据携带的 Watermark 比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。由于 Watermark 是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。
上图中，我们设置的允许最大延迟到达时间为 2s，所以时间戳为 7s 的事件对应的 Watermark 是 5s，时间戳为 12s 的事件的 Watermark 是 10s，如果我们的窗口 1 是 1s~5s，窗口 2 是 6s~10s，那么时间戳为 7s 的事件到达时的 Watermarker 恰好触发窗口 1，时间戳为 12s 的事件到达时的 Watermark 恰好触发窗口 2。
3) Flink 对于迟到数据的处理
waterMark 和 Window 机制解决了流式数据的乱序问题，对于因为延迟而顺序有误的数据，可以根据 eventTime 进行业务处理，于延迟的数据 Flink 也有自己的解决办法，主要的办法是给定一个允许延迟的时间，在该时间范围内仍可以接受处理延迟数据。
设置允许延迟的时间是通过 allowedLateness(lateness: Time) 设置
保存延迟数据则是通过 sideOutputLateData(outputTag: OutputTag[T]) 保存
获取延迟数据是通过 DataStream.getSideOutput(tag: OutputTag[X]) 获取
具体的用法如下：
allowedLateness(lateness: Time)
def allowedLateness(lateness: Time): WindowedStream[T, K, W] = { javaStream.allowedLateness(lateness) this }
该方法传入一个 Time 值，设置允许数据迟到的时间，这个时间和 WaterMark 中的时间概念不同。再来回顾一下：
WaterMark=数据的事件时间-允许乱序时间值
随着新数据的到来，waterMark 的值会更新为最新数据事件时间-允许乱序时间值，但是如果这时候来了一条历史数据，waterMark 值则不会更新。总的来说，waterMark 是为了能接收到尽可能多的乱序数据。
那这里的 Time 值，主要是为了等待迟到的数据，在一定时间范围内，如果属于该窗口的数据到来，仍会进行计算，后面会对计算方式仔细说明
注意：该方法只针对于基于 event-time 的窗口，如果是基于 processing-time，并且指定了非零的 time 值则会抛出异常。
sideOutputLateData(outputTag: OutputTag[T])
def sideOutputLateData(outputTag: OutputTag[T]): WindowedStream[T, K, W] = { javaStream.sideOutputLateData(outputTag) this }
该方法是将迟来的数据保存至给定的 outputTag 参数，而 OutputTag 则是用来标记延迟数据的一个对象。
DataStream.getSideOutput(tag: OutputTag[X])
通过 window 等操作返回的 DataStream 调用该方法，传入标记延迟数据的对象来获取延迟的数据。
对延迟数据的理解
延迟数据是指：
在当前窗口【假设窗口范围为 10-15】已经计算之后，又来了一个属于该窗口的数据【假设事件时间为 13】，这时候仍会触发 Window 操作，这种数据就称为延迟数据。
那么问题来了，延迟时间怎么计算呢？
假设窗口范围为 10-15，延迟时间为 2s，则只要 WaterMark<15+2，并且属于该窗口，就能触发 Window 操作。而如果来了一条数据使得 WaterMark>=15+2，10-15 这个窗口就不能再触发 Window 操作，即使新来的数据的 Event Time 属于这个窗口时间内 。
4) Flink 关联 Hive 分区表
Flink 1.12 支持了 Hive 最新的分区作为时态表的功能，可以通过 SQL 的方式直接关联 Hive 分区表的最新分区，并且会自动监听最新的 Hive 分区，当监控到新的分区后，会自动地做维表数据的全量替换。通过这种方式，用户无需编写 DataStream 程序即可完成 Kafka 流实时关联最新的 Hive 分区实现数据打宽。
具体用法：
在 Sql Client 中注册 HiveCatalog：
vim conf/sql-client-defaults.yaml
catalogs:
- name: hive_catalog
type: hive
hive-conf-dir: /disk0/soft/hive-conf/ #该目录需要包hive-site.xml文件 创建 Kafka 表
CREATE TABLE hive_catalog.flink_db.kfk_fact_bill_master_12 ( master Row<reportDate String, groupID int, shopID int, shopName String, action int, orderStatus int, orderKey String, actionTime bigint, areaName String, paidAmount double, foodAmount double, startTime String, person double, orderSubType int, checkoutTime String>, proctime as PROCTIME() -- PROCTIME用来和Hive时态表关联 ) WITH ( 'connector' = 'kafka', 'topic' = 'topic_name', 'format' = 'json', 'properties.bootstrap.servers' = 'host:9092', 'properties.group.id' = 'flinkTestGroup', 'scan.startup.mode' = 'timestamp', 'scan.startup.timestamp-millis' = '1607844694000' );
Flink 事实表与 Hive 最新分区数据关联
dim_extend_shop_info 是 Hive 中已存在的表，所以我们用 table hint 动态地开启维表参数。
CREATE VIEW IF NOT EXISTS hive_catalog.flink_db.view_fact_bill_master as SELECT * FROM (select t1.*, t2.group_id, t2.shop_id, t2.group_name, t2.shop_name, t2.brand_id, ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn from hive_catalog.flink_db.kfk_fact_bill_master_12 t1 JOIN hive_catalog.flink_db.dim_extend_shop_info /*+ OPTIONS('streaming-source.enable'='true', 'streaming-source.partition.include' = 'latest', 'streaming-source.monitor-interval' = '1 h', 'streaming-source.partition-order' = 'partition-name') */ FOR SYSTEM_TIME AS OF t1.proctime AS t2 --时态表 ON t1.groupID = t2.group_id and t1.shopID = t2.shop_id where groupID in (202042)) t where t.rn = 1
参数解释：
-
streaming-source.enable 开启流式读取 Hive 数据。
-
streaming-source.partition.include 有以下两个值：
- latest 属性: 只读取最新分区数据。
- all: 读取全量分区数据 ，默认值为 all，表示读所有分区，latest 只能用在 temporal join 中，用于读取最新分区作为维表，不能直接读取最新分区数据。
-
streaming-source.monitor-interval 监听新分区生成的时间、不宜过短 、最短是 1 个小时，因为目前的实现是每个 task 都会查询 metastore，高频的查可能会对 metastore 产生过大的压力。需要注意的是，1.12.1 放开了这个限制，但仍建议按照实际业务不要配个太短的 interval。
-
streaming-source.partition-order 分区策略，主要有以下 3 种，其中最为推荐的是 partition-name：
- partition-name 使用默认分区名称顺序加载最新分区
- create-time 使用分区文件创建时间顺序
- partition-time 使用分区时间顺序
六、Flink 状态管理
我们前面写的 wordcount 的例子，没有包含状态管理。如果一个 task 在处理过程中挂掉了，那么它在内存中的状态都会丢失，所有的数据都需要重新计算。从容错和消息处理的语义上(at least once, exactly once)，Flink 引入了 state 和 checkpoint。
因此可以说 flink 因为引入了 state 和 checkpoint 所以才支持的 exactly once
首先区分一下两个概念：
state：
state 一般指一个具体的 task/operator 的状态：
-
state 数据默认保存在 java 的堆内存中，TaskManage 节点的内存中。
-
operator 表示一些算子在运行的过程中会产生的一些中间结果。
checkpoint：
checkpoint 可以理解为 checkpoint 是把 state 数据定时持久化存储了，则表示了一个 Flink Job 在一个特定时刻的一份全局状态快照，即包含了所有 task/operator 的状态。
注意：task(subTask)是 Flink 中执行的基本单位。operator 指算子(transformation)
State 可以被记录，在失败的情况下数据还可以恢复。
Flink 中有两种基本类型的 State：
-
Keyed State
-
Operator State
Keyed State 和 Operator State，可以以两种形式存在：
-
原始状态(raw state)
-
托管状态(managed state)
托管状态是由 Flink 框架管理的状态。
我们说 operator 算子保存了数据的中间结果，中间结果保存在什么类型中，如果我们这里是托管状态，则由 flink 框架自行管理
原始状态由用户自行管理状态具体的数据结构，框架在做 checkpoint 的时候，使用 byte[]来读写状态内容，对其内部数据结构一无所知。
通常在 DataStream 上的状态推荐使用托管的状态，当实现一个用户自定义的 operator 时，会使用到原始状态。
1. State-Keyed State
基于 KeyedStream 上的状态。这个状态是跟特定的 key 绑定的，对 KeyedStream 流上的每一个 key，都对应一个 state，比如：stream.keyBy(…)。KeyBy 之后的 Operator State,可以理解为分区过的 Operator State。
保存 state 的数据结构：
ValueState：即类型为 T 的单值状态。这个状态与对应的 key 绑定，是最简单的状态了。它可以通过 update 方法更新状态值，通过 value()方法获取状态值。
ListState：即 key 上的状态值为一个列表。可以通过 add 方法往列表中附加值；也可以通过 get()方法返回一个 Iterable来遍历状态值。
ReducingState:这种状态通过用户传入的 reduceFunction，每次调用 add 方法添加值的时候，会调用 reduceFunction，最后合并到一个单一的状态值。
MapState<UK, UV>:即状态值为一个 map。用户通过 put 或 putAll 方法添加元素。
需要注意的是，以上所述的 State 对象，仅仅用于与状态进行交互（更新、删除、清空等），而真正的状态值，有可能是存在内存、磁盘、或者其他分布式存储系统中。相当于我们只是持有了这个状态的句柄。
1. ValueState
使用 ValueState 保存中间结果对下面数据进行分组求和。
开发步骤：
1. 获取流处理执行环境 2. 加载数据源 3. 数据分组 4. 数据转换，定义ValueState,保存中间结果 5. 数据打印 6. 触发执行
ValueState:测试数据源：
List( (1L, 4L), (2L, 3L), (3L, 1L), (1L, 2L), (3L, 2L), (1L, 2L), (2L, 2L), (2L, 9L) )
示例代码：
import org.apache.flink.api.common.functions.RichFlatMapFunction import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor} import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation} import org.apache.flink.configuration.Configuration import org.apache.flink.streaming.api.scala._ import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} import org.apache.flink.util.Collector object TestKeyedState { class CountWithKeyedState extends RichFlatMapFunction[(Long, Long), (Long, Long)] { /** * ValueState状态句柄. 第一个值为count，第二个值为sum。 */ private var sum: ValueState[(Long, Long)] = _ override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = { // 获取当前状态值 val tmpCurrentSum: (Long, Long) = sum.value // 状态默认值 val currentSum = if (tmpCurrentSum != null) { tmpCurrentSum } else { (0L, 0L) } // 更新 val newSum = (currentSum._1 + 1, currentSum._2 + input._2) // 更新状态值 sum.update(newSum) // 如果count >=3 清空状态值，重新计算 if (newSum._1 >= 3) { out.collect((input._1, newSum._2 / newSum._1)) sum.clear() } } override def open(parameters: Configuration): Unit = { sum = getRuntimeContext.getState( new ValueStateDescriptor[(Long, Long)]("average", // 状态名称 TypeInformation.of(new TypeHint[(Long, Long)](){}) )// 状态类型 ) } } def main(args: Array[String]): Unit = { //初始化执行环境 val env = StreamExecutionEnvironment.getExecutionEnvironment //构建数据源 val inputStream: DataStream[(Long, Long)] = env.fromCollection( List( (1L, 4L), (2L, 3L), (3L, 1L), (1L, 2L), (3L, 2L), (1L, 2L), (2L, 2L), (2L, 9L)) ) //执行数据处理 inputStream.keyBy(0) .flatMap(new CountWithKeyedState) .setParallelism(1) .print //运行任务 env.execute } }
2. MapState
使用 MapState 保存中间结果对下面数据进行分组求和:
1. 获取流处理执行环境 2. 加载数据源 3. 数据分组 4. 数据转换，定义MapState,保存中间结果 5. 数据打印 6. 触发执行
MapState:测试数据源：
List( ("java", 1), ("python", 3), ("java", 2), ("scala", 2), ("python", 1), ("java", 1), ("scala", 2) )
示例代码：
object MapState { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) /** * 使用MapState保存中间结果对下面数据进行分组求和 * 1.获取流处理执行环境 * 2.加载数据源 * 3.数据分组 * 4.数据转换，定义MapState,保存中间结果 * 5.数据打印 * 6.触发执行 */ val source: DataStream[(String, Int)] = env.fromCollection(List( ("java", 1), ("python", 3), ("java", 2), ("scala", 2), ("python", 1), ("java", 1), ("scala", 2))) source.keyBy(0) .map(new RichMapFunction[(String, Int), (String, Int)] { var mste: MapState[String, Int] = _ override def open(parameters: Configuration): Unit = { val msState = new MapStateDescriptor[String, Int]("ms", TypeInformation.of(new TypeHint[(String)] {}), TypeInformation.of(new TypeHint[(Int)] {})) mste = getRuntimeContext.getMapState(msState) } override def map(value: (String, Int)): (String, Int) = { val i: Int = mste.get(value._1) mste.put(value._1, value._2 + i) (value._1, value._2 + i) } }).print() env.execute() } }
2. State-Operator State
与 Key 无关的 State，与 Operator 绑定的 state，整个 operator 只对应一个 state。
保存 state 的数据结构：
ListState
举例来说，Flink 中的 Kafka Connector，就使用了 operator state。它会在每个 connector 实例中，保存该实例中消费 topic 的所有(partition, offset)映射。
步骤：
-
获取执行环境
-
设置检查点机制：路径，重启策略
-
自定义数据源
- 需要继承并行数据源和 CheckpointedFunction
- 设置 listState,通过上下文对象 context 获取
- 数据处理，保留 offset
- 制作快照
-
数据打印
-
触发执行
示例代码：
import java.util import org.apache.flink.api.common.restartstrategy.RestartStrategies import org.apache.flink.api.common.state.{ListState, ListStateDescriptor} import org.apache.flink.api.common.time.Time import org.apache.flink.api.common.typeinfo.{TypeHint, TypeInformation} import org.apache.flink.runtime.state.{FunctionInitializationContext, FunctionSnapshotContext} import org.apache.flink.runtime.state.filesystem.FsStateBackend import org.apache.flink.streaming.api.CheckpointingMode import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction import org.apache.flink.streaming.api.environment.CheckpointConfig import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction} import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment import org.apache.flink.streaming.api.scala._ object ListOperate { def main(args: Array[String]): Unit = { val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment env.setParallelism(1) env.enableCheckpointing(5000) env.setStateBackend(new FsStateBackend("hdfs://node01:8020/tmp/check/8")) env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) env.getCheckpointConfig.setMaxConcurrentCheckpoints(1) env.getCheckpointConfig.setCheckpointTimeout(60000) env.getCheckpointConfig.setFailOnCheckpointingErrors(false) env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION) //重启策略 env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(1), Time.seconds(5))) //模拟kakfa偏移量 env.addSource(new MyRichParrelSourceFun) .print() env.execute() } } class MyRichParrelSourceFun extends RichParallelSourceFunction[String] with CheckpointedFunction { var listState: ListState[Long] = _ var offset: Long = 0L //任务运行 override def run(ctx: SourceFunction.SourceContext[String]): Unit = { val iterState: util.Iterator[Long] = listState.get().iterator() while (iterState.hasNext) { offset = iterState.next() } while (true) { offset += 1 ctx.collect("offset:"+offset) Thread.sleep(1000) if(offset > 10){ 1/0 } } } //取消任务 override def cancel(): Unit = ??? //制作快照 override def snapshotState(context: FunctionSnapshotContext): Unit = { listState.clear() listState.add(offset) } //初始化状态 override def initializeState(context: FunctionInitializationContext): Unit = { listState = context.getOperatorStateStore.getListState(new ListStateDescriptor[Long]( "listState", TypeInformation.of(new TypeHint[Long] {}) )) } }
3. Broadcast State
Broadcast State 是 Flink 1.5 引入的新特性。在开发过程中，如果遇到需要下发/广播配置、规则等低吞吐事件流到下游所有 task 时，就可以使用 Broadcast State 特性。下游的 task 接收这些配置、规则并保存为 BroadcastState, 将这些配置应用到另一个数据流的计算中 。
1) API 介绍
通常，我们首先会创建一个 Keyed 或 Non-Keyed 的 Data Stream，然后再创建一个 Broadcasted Stream，最后通过 Data Stream 来连接（调用 connect 方法）到 Broadcasted Stream 上，这样实现将 Broadcast State 广播到 Data Stream 下游的每个 Task 中。
如果 Data Stream 是 Keyed Stream，则连接到 Broadcasted Stream 后，添加处理 ProcessFunction 时需要使用 KeyedBroadcastProcessFunction 来实现，下面是 KeyedBroadcastProcessFunction 的 API，代码如下所示：
public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> extends BaseBroadcastProcessFunction { public abstract void processElement(final IN1 value, final ReadOnlyContext ctx, final Collector<OUT> out) throws Exception; public abstract void processBroadcastElement(final IN2 value, final Context ctx, final Collector<OUT> out) throws Exception; }
上面泛型中的各个参数的含义，说明如下：
- KS：表示 Flink 程序从最上游的 Source Operator 开始构建 Stream，当调用 keyBy 时所依赖的 Key 的类型；
- IN1：表示非 Broadcast 的 Data Stream 中的数据记录的类型；
- IN2：表示 Broadcast Stream 中的数据记录的类型；
- OUT：表示经过 KeyedBroadcastProcessFunction 的 processElement()和 processBroadcastElement()方法处理后输出结果数据记录的类型。
如果 Data Stream 是 Non-Keyed Stream，则连接到 Broadcasted Stream 后，添加处理 ProcessFunction 时需要使用 BroadcastProcessFunction 来实现，下面是 BroadcastProcessFunction 的 API，代码如下所示：
public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction { public abstract void processElement(final IN1 value, final ReadOnlyContext ctx, final Collector<OUT> out) throws Exception; public abstract void processBroadcastElement(final IN2 value, final Context ctx, final Collector<OUT> out) throws Exception; }
上面泛型中的各个参数的含义，与前面 KeyedBroadcastProcessFunction 的泛型类型中的后 3 个含义相同，只是没有调用 keyBy 操作对原始 Stream 进行分区操作，就不需要 KS 泛型参数。
注意事项：
-
Broadcast State 是 Map 类型，即 K-V 类型。
-
Broadcast State 只有在广播一侧的方法中 processBroadcastElement 可以修改;在非广播一侧方法中 processElement 只读。
-
Broadcast State 在运行时保存在内存中。
2) 场景举例
-
动态更新计算规则: 如事件流需要根据最新的规则进行计算，则可将规则作为广播状态广播到下游 Task 中。
-
实时增加额外字段: 如事件流需要实时增加用户的基础信息，则可将用户的基础信息作为广播状态广播到下游 Task 中。
七、Flink 的容错
1. Checkpoint 介绍
checkpoint 机制是 Flink 可靠性的基石，可以保证 Flink 集群在某个算子因为某些原因(如 异常退出)出现故障时，能够将整个应用流图的状态恢复到故障之前的某一状态，保 证应用流图状态的一致性。Flink 的 checkpoint 机制原理来自“Chandy-Lamport algorithm”算法。
每个需要 checkpoint 的应用在启动时，Flink 的 JobManager 为其创建一个 CheckpointCoordinator(检查点协调器)，CheckpointCoordinator 全权负责本应用的快照制作。 
-
CheckpointCoordinator(检查点协调器) 周期性的向该流应用的所有 source 算子发送 barrier(屏障)。
-
当某个 source 算子收到一个 barrier 时，便暂停数据处理过程，然后将自己的当前状态制作成快照，并保存到指定的持久化存储中，最后向 CheckpointCoordinator 报告自己快照制作情况，同时向自身所有下游算子广播该 barrier，恢复数据处理
-
下游算子收到 barrier 之后，会暂停自己的数据处理过程，然后将自身的相关状态制作成快照，并保存到指定的持久化存储中，最后向 CheckpointCoordinator 报告自身快照情况，同时向自身所有下游算子广播该 barrier，恢复数据处理。
-
每个算子按照步骤 3 不断制作快照并向下游广播，直到最后 barrier 传递到 sink 算子，快照制作完成。
-
当 CheckpointCoordinator 收到所有算子的报告之后，认为该周期的快照制作成功; 否则，如果在规定的时间内没有收到所有算子的报告，则认为本周期快照制作失败。
如果一个算子有两个输入源，则暂时阻塞先收到 barrier 的输入源，等到第二个输入源相 同编号的 barrier 到来时，再制作自身快照并向下游广播该 barrier。具体如下图所示：
-
假设算子 C 有 A 和 B 两个输入源
-
在第 i 个快照周期中，由于某些原因(如处理时延、网络时延等)输入源 A 发出的 barrier 先到来，这时算子 C 暂时将输入源 A 的输入通道阻塞，仅收输入源 B 的数据。
-
当输入源 B 发出的 barrier 到来时，算子 C 制作自身快照并向 CheckpointCoordinator 报告自身的快照制作情况，然后将两个 barrier 合并为一个，向下游所有的算子广播。
-
当由于某些原因出现故障时，CheckpointCoordinator 通知流图上所有算子统一恢复到某个周期的 checkpoint 状态，然后恢复数据流处理。分布式 checkpoint 机制保证了数据仅被处理一次(Exactly Once)。
2. 持久化存储
。。。。。。。。。。。。。。
可关注公众号：五分钟学大数据，获取最新文章！本专栏专注于大数据领域，分享大数据技术文章，大数据面试技巧及大数据前沿技术！
投递科大讯飞等公司10个岗位 >
