Flink 系列文章

1、Flink1.12.7或1.13.5详细介绍及本地安装部署、验证
2、Flink1.13.5二种部署方式(Standalone、Standalone HA )、四种提交任务方式（前两种及session和per-job）验证详细步骤
3、flink重要概念（api分层、角色、执行流程、执行图和编程模型）及dataset、datastream详细示例入门和提交任务至on yarn运行
4、介绍Flink的流批一体、transformations的18种算子详细介绍、Flink与Kafka的source、sink介绍
5、Flink 的 source、transformations、sink的详细示例（一）
5、Flink的source、transformations、sink的详细示例（二）-source和transformation示例
5、Flink的source、transformations、sink的详细示例（三）-sink示例
6、Flink四大基石之Window详解与详细示例（一）
6、Flink四大基石之Window详解与详细示例（二）
7、Flink四大基石之Time和WaterMaker详解与详细示例（watermaker基本使用、kafka作为数据源的watermaker使用示例以及超出最大允许延迟数据的接收实现）
8、Flink四大基石之State概念、使用场景、持久化、批处理的详解与keyed state和operator state、broadcast state使用和详细示例
9、Flink四大基石之Checkpoint容错机制详解及示例（checkpoint配置、重启策略、手动恢复checkpoint和savepoint）
10、Flink的source、transformations、sink的详细示例（二）-source和transformation示例【补充示例】
11、Flink配置flink-conf.yaml详细说明（HA配置、checkpoint、web、安全、zookeeper、historyserver、workers、zoo.cfg）
12、Flink source和sink 的 clickhouse 详细示例
13、Flink 的table api与sql的基本概念、通用api介绍及入门示例
14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性
15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置（如何处理更新结果）、时态表、流上的join、流上的确定性以及查询配置

22、Flink 的table api与sql之创建表的DDL

30、Flink SQL之SQL 客户端（通过kafka和filesystem的例子介绍了配置文件使用-表、视图等）

本文是Flink系列的第一篇。
本文详细的介绍了flink的背景与优势、应用场景，同时介绍了flink的本地部署及验证、提交任务等。
本文部分图片来源于互联网。
本文依赖jdk环境可用，需要jdk 8及以后的版本。
本文分为2个部分，及flink介绍及本地化部署、提交任务、验证。

一、Flink介绍

官网：http://iyenn.com/index/link?url=https://flink.apache.org/
官网中文：http://iyenn.com/index/link?url=http://iyenn.com/index/link?url=https://flink.apache.org/zh/
本示例以1.12版本进行介绍，当前版本更新至1.17。

1、1.12版本重要更新

在 DataStream API 上添加了高效的批执行模式的支持。这是批处理和流处理实现真正统一的运行时的一个重要里程碑
实现了基于Kubernetes的高可用性（HA）方案，作为生产环境中，ZooKeeper方案之外的另外一种选择
扩展了 Kafka SQL connector，使其可以在 upsert 模式下工作，并且支持在 SQL DDL 中处理 connector 的 metadata。现在，时态表 Join 可以完全用 SQL 来表示，不再依赖于 Table API 了
PyFlink 中添加了对于 DataStream API 的支持，将 PyFlink 扩展到了更复杂的场景，比如需要对状态或者定时器 timer 进行细粒度控制的场景。除此之外，现在原生支持将 PyFlink 作业部署到 Kubernetes上

2、flink介绍

Flink 诞生于欧洲的一个大数据研究项目 StratoSphere。该项目是柏林工业大学的一个研究性项目。早期， Flink 是做 Batch 计算的，但是在 2014 年， StratoSphere 里面的核心成员孵化出 Flink，同年将 Flink 捐赠 Apache，并在后来成为 Apache 的顶级大数据项目，同时 Flink 计算的主流方向被定位为 Streaming， 即用流式计算来做所有大数据的计算。

2014 年 Flink 作为主攻流计算的大数据引擎开始在开源大数据行业内崭露头角。区别于 Storm、Spark Streaming 以及其他流式计算引擎的是：它不仅是一个高吞吐、低延迟的计算引擎，同时还提供很多高级的功能。比如它提供了有状态的计算，支持状态管理，支持强一致性的数据语义以及支持 基于Event Time的WaterMark对延迟或乱序的数据进行处理等。

2015 年阿里巴巴开始使用 Flink 并持续贡献社区(阿里内部还基于Flink做了一套Blink)，2019年1月8日，阿里巴巴以 9000 万欧元(7亿元人民币)收购了创业公司 Data Artisans。

3、编程语言

Flink官方提供了Java、Scala、Python语言接口用以开发Flink应用程序，但是Flink的源码是使用Java语言进行开发的，且Flink被阿里收购后，未来的主要编程语言可能主要会是Java，且GitHub上关于Flink的项目，大多数是使用Java语言编写的。

4、组件介绍

1）、Deploy物理部署层

Flink 支持本地运行、能在独立集群或者在被 YARN 管理的集群上运行， 也能部署在云上。
该层主要涉及Flink的部署模式，目前Flink支持多种部署模式，包括本地、集群(Standalone、YARN)、云(GCE/EC2)、Kubenetes。
Flink能够通过该层能够支持不同平台的部署，用户可以根据需要选择使用对应的部署模式。

2）、Core Runtime核心层

Runtime层提供了支持Flink计算的全部核心实现，为上层API层提供基础服务，该层主要负责对上层不同接口提供基础服务，也是Flink分布式计算框架的核心实现层。
支持分布式Stream作业的执行、JobGraph到ExecutionGraph的映射转换、任务调度等。
将DataSteam和DataSet转成统一的可执行的Task Operator，达到在流式引擎下同时处理批量计算和流式计算的目的。

3）、API&Libraries层

Flink 首先支持了 Scala 和 Java 的 API、Python 。
DataStream、DataSet、Table、SQL API，作为分布式数据处理框架，Flink同时提供了支撑计算和批计算的接口，两者都提供给用户丰富的数据处理高级API，例如Map、FlatMap操作等，也提供比较低级的Process Function API，用户可以直接操作状态和时间等底层数据。

4）、扩展库

Flink 还包括用于复杂事件处理的CEP，机器学习库FlinkML，图处理库Gelly等。
Table 是一种接口化的 SQL 支持，也就是 API 支持(DSL)，而不是文本化的SQL 解析和执行。

5、四大基石

1）、Checkpoint

这是Flink最重要的一个特性。
Flink基于Chandy-Lamport算法实现了一个分布式的一致性的快照，从而提供了一致性的语义。

2）、State

提供了一致性的语义之后，Flink为了让用户在编程时能够更轻松、更容易地去管理状态，还提供了一套非常简单明了的State API，包括里面的有ValueState、ListState、MapState、BroadcastState，使用State API能够自动享受到这种一致性的语义。

3）、Time

Flink实现了Watermark的机制，能够支持基于事件的时间的处理，能够容忍迟到/乱序的数据。

4）、Window

流计算中一般在对流数据进行操作之前都会先进行开窗，即基于一个什么样的窗口上做这个计算。Flink提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义的窗口

6、应用场景

Apache Flink 功能强大，支持开发和运行多种不同种类的应用程序。它的主要特性包括：批流一体化、精密的状态管理、事件时间支持以及精确一次的状态一致性保障等。
Flink 不仅可以运行在包括 YARN、 Mesos、Kubernetes 在内的多种资源管理框架上，还支持在裸机集群上独立部署。
在启用高可用选项的情况下，它不存在单点失效问题。事实证明，Flink 已经可以扩展到数千核心，其状态可以达到 TB 级别，且仍能保持高吞吐、低延迟的特性。
世界各地有很多要求严苛的流处理应用都运行在 Flink 之上。

1）、事件驱动型应用

事件驱动型应用是一类具有状态的应用，它从一个或多个事件流提取数据，并根据到来的事件触发计算、状态更新或其他外部动作。
事件驱动型应用是在计算存储分离的传统应用基础上进化而来。在传统架构中，应用需要读写远程事务型数据库。

相反，事件驱动型应用是基于状态化流处理来完成。在该设计中，数据和计算不会分离，应用只需访问本地（内存或磁盘）即可获取数据。系统容错性的实现依赖于定期向远程持久化存储写入 checkpoint。下图描述了传统应用和事件驱动型应用架构的区别。

事件驱动型应用会受制于底层流处理系统对时间和状态的把控能力，Flink 诸多优秀特质都是围绕这些方面来设计的。它提供了一系列丰富的状态操作原语，允许以精确一次的一致性语义合并海量规模（TB 级别）的状态数据。此外，Flink 还支持事件时间和自由度极高的定制化窗口逻辑，而且它内置的 ProcessFunction 支持细粒度时间控制，方便实现一些高级业务逻辑。同时，Flink 还拥有一个复杂事件处理（CEP）类库，可以用来检测数据流中的模式。

Flink 中针对事件驱动应用的明星特性当属 savepoint。Savepoint 是一个一致性的状态映像，它可以用来初始化任意状态兼容的应用。在完成一次 savepoint 后，即可放心对应用升级或扩容，还可以启动多个版本的应用来完成 A/B 测试。

典型事件驱动应用实例：

• 欺诈检测(Fraud detection)
• 异常检测(Anomaly detection)
• 基于规则的告警(Rule-based alerting)
• 业务流程监控(Business process monitoring)
• Web应用程序(社交网络)

2）、数据分析应用

数据分析任务需要从原始数据中提取有价值的信息和指标。传统的分析方式通常是利用批查询，或将事件记录下来并基于此有限数据集构建应用来完成。为了得到最新数据的分析结果，必须先将它们加入分析数据集并重新执行查询或运行应用，随后将结果写入存储系统或生成报告。
借助一些先进的流处理引擎，还可以实时地进行数据分析。和传统模式下读取有限数据集不同，流式查询或应用会接入实时事件流，并随着事件消费持续产生和更新结果。这些结果数据可能会写入外部数据库系统或以内部状态的形式维护。仪表展示应用可以相应地从外部数据库读取数据或直接查询应用的内部状态。
如下图所示，Apache Flink 同时支持流式及批量分析应用。

Flink 为持续流式分析和批量分析都提供了良好的支持。具体而言，它内置了一个符合 ANSI 标准的 SQL 接口，将批、流查询的语义统一起来。无论是在记录事件的静态数据集上还是实时事件流上，相同 SQL 查询都会得到一致的结果。同时 Flink 还支持丰富的用户自定义函数，允许在 SQL 中执行定制化代码。如果还需进一步定制逻辑，可以利用 Flink DataStream API 和 DataSet API 进行更低层次的控制。

典型的数据分析应用实例

• 电信网络质量监控
• 移动应用中的产品更新及实验评估分析
• 消费者技术中的实时数据即席分析
• 大规模图分析

3）、数据管道应用

提取-转换-加载（ETL）是一种在存储系统之间进行数据转换和迁移的常用方法。ETL 作业通常会周期性地触发，将数据从事务型数据库拷贝到分析型数据库或数据仓库。
数据管道和 ETL 作业的用途相似，都可以转换、丰富数据，并将其从某个存储系统移动到另一个。但数据管道是以持续流模式运行，而非周期性触发。因此它支持从一个不断生成数据的源头读取记录，并将它们以低延迟移动到终点。例如：数据管道可以用来监控文件系统目录中的新文件，并将其数据写入事件日志；另一个应用可能会将事件流物化到数据库或增量构建和优化查询索引。
下图描述了周期性 ETL 作业和持续数据管道的差异。

很多常见的数据转换和增强操作可以利用 Flink 的 SQL 接口（或 Table API）及用户自定义函数解决。如果数据管道有更高级的需求，可以选择更通用的 DataStream API 来实现。Flink 为多种数据存储系统（如：Kafka、Kinesis、Elasticsearch、JDBC数据库系统等）内置了连接器。同时它还提供了文件系统的连续型数据源及数据汇，可用来监控目录变化和以时间分区的方式写入文件。

典型的数据管道应用实例

• 电子商务中的实时查询索引构建
• 电子商务中的持续 ETL

7、flink优势

1. Flink 具备统一的框架处理有界和无界两种数据流的能力

2. 部署灵活，Flink 底层支持多种资源调度器，包括Yarn、Kubernetes 等。Flink 自身带的Standalone 的调度器，在部署上也十分灵活。

3. 极高的可伸缩性，可伸缩性对于分布式系统十分重要。

4. 极致的流式处理性能。Flink 相对于Storm 最大的特点是将状态语义完全抽象到框架中，支持本地状态读取，避免了大量网络IO，可以极大提升状态存取的性能。

5. 同时支持高吞吐、低延迟、高性能。Flink 是目前开源社区中唯一一套集高吞吐、低延迟、高性能三者于一身的分布式流式数据处理框架。Spark 只能兼顾高吞吐和高性能特性，无法做到低延迟保障，因为Spark是用批处理来做流处理。Storm 只能支持低延时和高性能特性，无法满足高吞吐的要求。
   下图显示了 Apache Flink 与 Apache Storm 在完成流数据清洗的分布式任务的性能对比。
   

6. 支持事件时间(Event Time)概念。在流式计算领域中，窗口计算的地位举足轻重，但目前大多数框架窗口计算采用的都是系统时间(Process Time)，也就是事件传输到计算框架处理时，系统主机的当前时间。Flink 能够支持基于事件时间(Event Time)语义进行窗口计算。这种基于事件驱动的机制使得事件即使乱序到达甚至延迟到达，流系统也能够计算出精确的结果，保持了事件原本产生时的时序性，尽可能避免网络传输或硬件系统的影响。
   

7. 支持有状态计算。Flink1.4开始支持有状态计算。所谓状态就是在流式计算过程中将算子的中间结果保存在内存或者文件系统中，等下一个事件进入算子后可以从之前的状态中获取中间结果，计算当前的结果，从而无须每次都基于全部的原始数据来统计结果，极大的提升了系统性能，状态化意味着应用可以维护随着时间推移已经产生的数据聚合。
   

8. 支持高度灵活的窗口(Window)操作。Flink 将窗口划分为基于 Time 、Count 、Session、以及Data-Driven等类型的窗口操作，窗口可以用灵活的触发条件定制化来达到对复杂的流传输模式的支持，用户可以定义不同的窗口触发机制来满足不同的需求。

9. 基于轻量级分布式快照(Snapshot/Checkpoints)的容错机制。Flink 能够分布运行在上千个节点上，通过基于分布式快照技术的Checkpoints，将执行过程中的状态信息进行持久化存储，一旦任务出现异常停止，Flink 能够从 Checkpoints 中进行任务的自动恢复，以确保数据处理过程中的一致性。Flink 的容错能力是轻量级的，允许系统保持高并发，同时在相同时间内提供强一致性保证。
   

10. 基于 JVM 实现的独立的内存管理。Flink 实现了自身管理内存的机制，通过使用散列，索引，缓存和排序有效地进行内存管理，通过序列化/反序列化机制将所有的数据对象转换成二进制在内存中存储，降低数据存储大小的同时，更加有效的利用空间。使其独立于 Java 的默认垃圾收集器，尽可能减少 JVM GC 对系统的影响。

11. SavePoints。对于 7 * 24 小时运行的流式应用，数据源源不断的流入，在一段时间内应用的终止有可能导致数据的丢失或者计算结果的不准确。比如集群版本的升级，停机运维操作等。Flink 通过SavePoints 技术将任务执行的快照保存在存储介质上，当任务重启的时候，可以从事先保存的 SavePoints 恢复原有的计算状态，使得任务继续按照停机之前的状态运行。Flink 保存点提供了一个状态化的版本机制，使得能以无丢失状态和最短停机时间的方式更新应用或者回退历史数据。
    

12. 灵活的部署方式，支持大规模集群。Flink 被设计成能用上千个点在大规模集群上运行。除了支持独立集群部署外，Flink 还支持 YARN 和Mesos 方式部署。

13. Flink 的程序内在是并行和分布式的。数据流可以被分区成 stream partitions，operators 被划分为operator subtasks；这些 subtasks 在不同的机器或容器中分不同的线程独立运行；operator subtasks 的数量就是operator的并行计算数，不同的 operator 阶段可能有不同的并行数；如下图所示，source operator 的并行数为 2，但最后的 sink operator 为1。
    

14. 丰富的库。Flink 拥有丰富的库来进行机器学习，图形处理，关系数据处理等。

二、本地部署及验证

Flink支持多种安装模式

• Local—本地单机模式，学习测试时使用
• Standalone—独立集群模式，Flink自带集群，开发测试环境使用
• StandaloneHA—独立集群高可用模式，Flink自带集群，开发测试环境使用
• On Yarn—计算资源统一由Hadoop YARN管理，生产环境使用
  部署前提，最新版本1.17要求java 11以上版本，1.12还可以使用java 8版本。

1、本地单机部署

1）、下载安装文件

下载地址：http://iyenn.com/index/link?url=http://iyenn.com/index/link?url=https://flink.apache.org/downloads/
选择自己需要的版本下载即可

本示例选择的是1.13.5版本

2）、解压

tar -xzf flink-1.13.5-bin-scala_2.11.tgz  -C /usr/local/flink-1.13.5
cd /usr/local/flink-1.13.5/bin

[root@server2 bin]# pwd
/usr/local/flink-1.13.5/bin
[root@server2 bin]# ll
总用量 2368
-rw-r--r-- 1 1006 1007 2289670 12月 15 2021 bash-java-utils.jar
-rwxr-xr-x 1 1006 1007   20576 12月 14 2021 config.sh
-rwxr-xr-x 1 1006 1007    1318 12月 14 2021 find-flink-home.sh
-rwxr-xr-x 1 1006 1007    2381 12月 14 2021 flink
-rwxr-xr-x 1 1006 1007    4137 12月 14 2021 flink-console.sh
-rwxr-xr-x 1 1006 1007    6584 12月 14 2021 flink-daemon.sh
-rwxr-xr-x 1 1006 1007    1564 12月 14 2021 historyserver.sh
-rwxr-xr-x 1 1006 1007    2295 12月 14 2021 jobmanager.sh
-rwxr-xr-x 1 1006 1007    1650 12月 14 2021 kubernetes-jobmanager.sh
-rwxr-xr-x 1 1006 1007    1717 12月 14 2021 kubernetes-session.sh
-rwxr-xr-x 1 1006 1007    1770 12月 14 2021 kubernetes-taskmanager.sh
-rwxr-xr-x 1 1006 1007    1133 12月 14 2021 mesos-appmaster-job.sh
-rwxr-xr-x 1 1006 1007    1137 12月 14 2021 mesos-appmaster.sh
-rwxr-xr-x 1 1006 1007    1958 12月 14 2021 mesos-jobmanager.sh
-rwxr-xr-x 1 1006 1007    1891 12月 14 2021 mesos-taskmanager.sh
-rwxr-xr-x 1 1006 1007    2994 12月 14 2021 pyflink-shell.sh
-rwxr-xr-x 1 1006 1007    3742 12月 14 2021 sql-client.sh
-rwxr-xr-x 1 1006 1007    2006 12月 14 2021 standalone-job.sh
-rwxr-xr-x 1 1006 1007    1837 12月 14 2021 start-cluster.sh
-rwxr-xr-x 1 1006 1007    3380 12月 14 2021 start-scala-shell.sh
-rwxr-xr-x 1 1006 1007    1854 12月 14 2021 start-zookeeper-quorum.sh
-rwxr-xr-x 1 1006 1007    1617 12月 14 2021 stop-cluster.sh
-rwxr-xr-x 1 1006 1007    1845 12月 14 2021 stop-zookeeper-quorum.sh
-rwxr-xr-x 1 1006 1007    2960 12月 14 2021 taskmanager.sh
-rwxr-xr-x 1 1006 1007    1725 12月 14 2021 yarn-session.sh
-rwxr-xr-x 1 1006 1007    2405 12月 14 2021 zookeeper.sh
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

3）、启动集群

Flink 附带了一个 bash 脚本，可以用于启动本地集群。

start-cluster.sh

[root@server2 bin]# start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host server2.
Starting taskexecutor daemon on host server2.
[root@server2 bin]# jps
26737 Jps
25707 StandaloneSessionClusterEntrypoint
26015 TaskManagerRunner
1
2
3
4
5
6
7
8
9
10

4）、提交作业

# 提交任务
flink run ../examples/streaming/WordCount.jar
# 查看运行结果
tail flink-*-taskexecutor-*.out

[root@server2 bin]# flink run ../examples/streaming/WordCount.jar
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
Job has been submitted with JobID c9fd6c83210c067f59ea63ad2b8b2ed1
Program execution finished
Job with JobID c9fd6c83210c067f59ea63ad2b8b2ed1 has finished.
Job Runtime: 718 ms

[root@server2 log]# pwd
/usr/local/flink-1.13.5/log
[root@server2 log]# tail flink-*-taskexecutor-*.out
==> flink-root-taskexecutor-0-localhost.localdomain.out <==

==> flink-root-taskexecutor-0-server2.out <==
(nymph,1)
(in,3)
(thy,1)
(orisons,1)
(be,4)
(all,2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

也可以通过 Flink 的 Web UI 来监视集群的状态和正在运行的作业，UI界面如下：


查看运行结果

5）、关闭集群

stop-cluster.sh

[root@server2 bin]# stop-cluster.sh
Stopping taskexecutor daemon (pid: 26015) on host server2.
Stopping standalonesession daemon (pid: 25707) on host server2.
1
2
3
4
5

以上，详细的介绍了flink的背景与优势、应用场景，同时介绍了flink的本地部署及验证、提交任务等。