第一章 初识Flink

大数据开发总体架构

大数据开发总体架构

数据传输层：
常用的数据传输工具有Flume、Sqoop、Kafka。Flume是一个日志收集系统，用于将大量日志数据从不同的源进行收集、聚合，最终移动到一个集中的数据中心进行存储。Sqoop主要用于将数据在关系型数据库和Hadoop平台之间进行相互转移。Kafka是一个发布与订阅消息系统，它可以实时处理大量消息数据以满足各种需求，相当于数据中转站。

数据存储层：
数据可以存储于分布式文件系统HDFS中，也可以存储于分布式数据库HBase中，而HBase的底层实际上还是将数据存储于HDFS中。此外，为了满足对大量数据的快速检索与统计，可以使用Elasticsearch作为全文检索引擎。

资源管理层：
YARN是大数据开发中常用的资源管理器，它是一个通用资源（内存、CPU）管理系统，不仅可以集成于Hadoop中，也可以集成于Flink、Spark等其他大数据框架中。

数据计算层：
MapReduce是Hadoop的核心组成部分，可以结合Hive通过SQL的方式进行数据的离线计算，当然也可以单独编写MapReduce应用程序进行计算。Storm用于进行数据的实时计算，可以非常容易地实时处理无限的流数据。Flink提供了离线计算库和实时计算库两种，离线计算库支持FlinkML（机器学习）、Gelly（图计算）、基于Table的关系操作，实时计算库支持CEP（复杂事件处理），同时也支持基于Table的关系操作。

任务调度层：
Oozie是一个用于Hadoop平台的工作流调度引擎，可以使用工作流的方式对编写好的大数据任务进行调度。若任务不复杂，则可以使用Linux系统自带的Crontab定时任务进行调度。

业务模型层：
对大量数据的处理结果最终需要通过可视化的方式进行展示。可以使用Java、PHP等处理业务逻辑，查询结果数据库，最终结合ECharts等前端可视化框架展示处理结果。

从另一个角度理解Flink在大数据开发架构中的位置，如图。

Flink在大数据开发架构中的位置

什么是Flink

Apache Flink是一个框架和分布式处理引擎，用于对无边界和有边界的数据流进行有状态的计算。Flink被设计为可以在所有常见集群环境中运行，并能以内存速度和任意规模执行计算。目前市场上主流的流式计算框架有Apache Storm、Spark Streaming、Apache Flink等，但能够同时支持低延迟、高吞吐、Exactly-Once（收到的消息仅处理一次）的框架只有Apache Flink。

Flink是原生的流处理系统，但也提供了批处理API，拥有基于流式计算引擎处理批量数据的计算能力，真正实现了批流统一。与Spark批处理不同的是，Flink把批处理当作流处理中的一种特殊情况。在Flink中，所有的数据都看作流，是一种很好的抽象，因为这更接近于现实世界。

Flink的主要优势如下：

1. 同时支持高吞吐、低延迟
   Flink是目前开源社区中唯一同时支持高吞吐、低延迟的分布式流式数据处理框架，在每秒处理数百万条事件的同时能够保持毫秒级延迟。而同类框架Spark Streaming在流式计算中无法做到低延迟保障。Apache Storm可以做到低延迟，但无法满足高吞吐的要求。同时满足高吞吐、低延迟对流式数据处理框架是非常重要的，可以大大提高数据处理的性能。

2. 支持有状态计算
   所谓状态，就是在流式计算过程中将算子（Flink提供了丰富的用于数据处理的函数，这些函数称为算子）的中间结果（需要持续聚合计算，依赖后续的数据记录）保存在内存或者文件系统中，等下一个事件进入算子后可以从之前的状态中获取中间结果，以便计算当前的结果（当前结果的计算可能依赖于之前的中间结果），从而无须每次都基于全部的原始数据来统计结果，极大地提升了系统性能。

   
   
   支持有状态计算

3. 支持事件时间
   时间是流处理框架的一个重要组成部分。目前大多数框架计算采用的都是系统处理时间（Process Time），也就是事件传输到计算框架处理时，系统主机的当前时间。Flink除了支持处理时间外，还支持事件时间（Event Time），根据事件本身自带的时间戳（事件的产生时间）进行结果的计算，例如窗口聚合、会话计算、模式检测和基于时间的聚合等。这种基于事件驱动的机制使得事件即使乱序到达，Flink也能够计算出精确的结果，保证了结果的准确性和一致性。

4. 支持高可用性配置
   Flink可以与YARN、HDFS、ZooKeeper等紧密集成，配置高可用，从而可以实现快速故障恢复、动态扩容、7×24小时运行流式应用等作业。Flink可以将任务执行的快照保存在存储介质上，当需要停机运维等操作时，下次启动可以直接从事先保存的快照恢复原有的计算状态，使得任务继续按照停机之前的状态运行。

5. 提供了不同层级的API
   Flink为流处理和批处理提供了不同层级的API，每一种API在简洁性和表达力上有着不同的侧重，并且针对不同的应用场景，不同层级的API降低了系统耦合度，也为用户构建Flink应用程序提供了丰富且友好的接口。

Flink的应用场景

1. 事件驱动
   根据到来的事件流触发计算、状态更新或其他外部动作，主要应用实例有反欺诈、异常检测、基于规则的报警、业务流程监控、（社交网络）Web应用等。
   传统应用和事件驱动型应用架构的区别如图：

传统架构和事件驱动架构对比

2. 数据分析
   从原始数据中提取有价值的信息和指标，这些信息和指标数据可以写入外部数据库系统或以内部状态的形式维护，主要应用实例有电信网络质量监控、移动应用中的产品更新及实验评估分析、实时数据分析、大规模图分析等。
   Flink同时支持批量及流式分析应用，如图

Flink支持批量和流式分析

3. 数据管道
   数据管道和ETL（Extract-Transform-Load，提取-转换-加载）作业的用途相似，都可以转换、丰富数据，并将其从某个存储系统移动到另一个。与ETL不同的是，ETL作业通常会周期性地触发，将数据从事务型数据库复制到分析型数据库或数据仓库。但数据管道是以持续流模式运行的，而非周期性触发，它支持从一个不断生成数据的源头读取记录，并将它们以低延迟移动到终点。例如，监控文件系统目录中的新文件，并将其数据写入事件日志。

数据管道的主要应用实例有电子商务中的实时查询索引构建、持续ETL等。周期性ETL作业和持续数据管道的对比如图：

Flink支持数据管道

流计算框架对比

当前大数据领域主流的流式计算框架有Apache Storm、Spark Streaming、Apache Flink三种。通常将Apache Storm称为第一代流式计算框架，Spark Streaming称为第二代流式计算框架，现在又出现了一种优秀的第三代实时计算框架Apache Flink，这三种计算框架的区别如表：

Apache Storm、Spark Streaming、Apache Flink对比

1. 模型
   Native：原生流处理。指输入的数据一旦到达，就立即进行处理，一次处理一条数据，如图

   
   
   流处理

Micro-Batching：微批流处理。把输入的数据按照预先定义的时间间隔（例如1秒钟）分成短小的批量数据，流经流处理系统进行处理，如图

批处理

Storm和Flink使用的是原生流处理，一次处理一条数据，是真正意义的流处理；而Spark Streaming实际上是通过批处理的方式模拟流处理，一次处理一批数据（小批量）。

2. API
   Storm只提供了组合式的基础API；而Spark Streaming和Flink都提供了封装后的高阶函数，例如map()、filter()，以及一些窗口函数、聚合函数等，使用这些函数可以轻松处理复杂的数据，构建并行应用程序。

3. 处理次数
   在流处理系统中，对数据的处理有3种级别的语义：At-Most-Once（最多一次）、At-Least-Once（至少一次）、Exactly-Once（仅一次）。
   由此可见，衡量一个流处理系统能力的关键是Exactly-Once。

4. 容错
   Storm通过使用ACK（确认回执，即数据接收方接收到数据后要向发送方发送确认回执，以此来保证数据不丢失）机制来确认每一条数据是否被成功处理，当处理失败时，则重新发送数据。这样很容易做到保证所有数据均被处理，没有遗漏，但这种方式不能保证数据仅被处理一次，因此存在同一条数据重复处理的情况。
   由于Spark Streaming是微批处理，不是真正意义上的流处理，其容错机制的实现相对简单。Spark Streaming中的每一批数据成为一个RDD（Resilient Distributed Dataset，分布式数据集）。RDD Checkpoint（检查点）机制相当于对RDD数据进行快照，可以将经常使用的RDD快照到指定的文件系统中，例如HDFS。当机器发生故障导致内存或磁盘中的RDD数据丢失时，可以快速从快照中对指定的RDD进行恢复。
   Flink的容错机制是基于分布式快照实现的，通过CheckPoint机制保存流处理作业某些时刻的状态，当任务异常结束时，默认从最近一次保存的完整快照处恢复任务。

5. 状态
   流处理系统的状态管理是非常重要的，Storm没有实现状态管理，Spark Streaming和Flink都实现了状态管理。通过状态管理可以把程序运行中某一时刻的数据结果保存起来，以便于后续的计算和故障的恢复。

6. 延迟
   由于Storm和Flink是接收到一条数据就立即处理，因此数据处理的延迟很低；而Spark Streaming是微批处理，需要形成一小批数据才会处理，数据处理的延迟相对偏高。

7. 吞吐量
   Storm的吞吐量相对来说较低，Spark Streaming和Flink的吞吐量则比较高。较高的吞吐量可以提高资源利用率，减小系统开销。

总的来说，Storm非常适合任务量小且延迟要求低的应用，但要注意Storm的容错恢复和状态管理都会降低整体的性能水平。如果你要使用Lambda架构，并且要集成Spark的各种库，那么Spark Streaming是一个不错的选择，但是要注意微批处理的局限性以及延迟问题。Flink可以满足绝大多数流处理场景，提供了丰富的高阶函数，并且也针对批处理场景提供了相应的API，是非常有前景的一个项目。

Flink主要组件

Flink是由多个组件构成的软件栈，整个软件栈可分为4层，如图：

Flink软件栈

（1）存储层
Flink本身并没有提供分布式文件系统，因此Flink的分析大多依赖于HDFS，也可以从HBase和Amazon S3（亚马逊云存储服务）等持久层读取数据。

（2）调度层
Flink自带一个简易的资源调度器，称为独立调度器（Standalone）。若集群中没有任何资源管理器，则可以使用自带的独立调度器。当然，Flink也支持在其他的集群管理器上运行，包括Hadoop YARN、Apache Mesos等。

（3）计算层
Flink的核心是一个对由很多计算任务组成的、运行在多个工作机器或者一个计算集群上的应用进行调度、分发以及监控的计算引擎，为API工具层提供基础服务。

（4）工具层
在Flink Runtime的基础上，Flink提供了面向流处理（DataStream API）和批处理（DataSet API）的不同计算接口，并在此接口上抽象出了不同的应用类型组件库，例如基于流处理的CEP（复杂事件处理库）、Table&SQL（结构化表处理库）和基于批处理的Gelly（图计算库）、FlinkML（机器学习库）、Table&SQL（结构化表处理库）。

Flink编程模型——数据集

在Flink的世界观中，任何类型的数据都可以形成一种事件流。例如信用卡交易、传感器测量、服务器日志、网站或移动应用程序上的用户交互记录等，所有这些数据都可以形成一种流，因为数据都是一条一条产生的。

根据数据流是否有时间边界，可将数据流分为有界流和无界流。有界流产生的数据集称为有界数据集，无界流产生的数据集称为无界数据集，如图

Flink有界数据集和无界数据集

1. 有界数据集
   定义一个数据流的开始，也定义数据流的结束，就会产生有界数据集。有界数据集的特点是数据是静止不动的，或者说当处理此类数据时不考虑数据的追加操作。例如，读取MySQL数据库、文本文件、HDFS系统等存储介质中的数据进行计算分析。

有界数据集具有时间边界，时间范围可能是一分钟，也可能是一天内的交易数据。可以在读取所有数据后再进行计算，对有界数据集的处理通常称为批处理（Batch Processing）。

批处理的数据查询方式如图：

Flink有界数据集

2. 无界数据集
   定义一个数据流的开始，但没有定义数据流的结束，就会产生无界数据集。无界数据集会无休止地产生新数据，是没有边界的。例如，实时读取Kafka中的消息数据进行计算、实时日志监控等。

对无界数据集必须持续处理，即数据被读取后需要立刻处理，不能等到所有数据都到达再处理，因为数据输入是无限的，在任何时候输入都不会完成。处理无界数据集通常要求以特定顺序读取事件（例如事件发生的顺序），以便能够推断结果的完整性。对无界数据集的处理被称为流处理。

有界数据集与无界数据集其实是一个相对的概念，如果每间隔一分钟、一小时、一天对数据进行一次计算，那么认为这一段时间的数据相对是有界的。有界的流数据又可以一条一条地按照顺序发送给计算引擎进行处理，在这种情况下可以认为数据是相对无界的。因此，有界数据集与无界数据集可以相互转换。Flink正是使用这种方式将有界数据集与无界数据集进行统一处理，从而将批处理和流处理统一在一套流式引擎中，能够同时实现批处理与流处理任务。

流处理数据查询方式如图：

Flink流处理查询

Flink编程模型——编程接口

Flink提供了丰富的数据处理接口，并将接口抽象成4层，由下向上分别为Stateful Stream Processing API、DataStream/DataSet API、Table API以及SQL API，开发者可以根据具体需求选择任意一层接口进行应用开发,如图：

Flink编程模型

1. Stateful Stream Processing API
   Flink中处理有状态流最底层的接口，使用Stateful Stream Process API接口可以实现非常复杂的流式计算逻辑，开发灵活性非常强，但是用户使用成本也相对较高。

2. DataStream/DataSet API
   实际上，大多数应用程序不需要上述低级抽象，而是针对核心API进行编程的，例如DataStream API和DataSet API。DataStream API用于处理无界数据集，即流处理；DataSet API用于处理有界数据集，即批处理。这两种API都提供了用于数据处理的通用操作，例如各种形式的转换、连接、聚合等。

3. Table API
   Table API作为批处理和流处理统一的关系型API，即查询在无界实时流或有界批数据集上以相同的语义执行，并产生相同的结果。
   Table API构建在DataStream/DataSet API之上，提供了大量编程接口，例如GroupByKey、Join等操作，是批处理和流处理统一的关系型API，使用起来更加简洁。使用Table API允许在表与DataStream/DataSet数据集之间无缝切换，并且可以将Table API与DataStream/DataSet API混合使用。

4. SQL API
   Flink提供的最高级别的抽象是SQL API。这种抽象在语义和表达方式上均类似于Table API，但是将程序表示为SQL查询表达式。SQL抽象与Table API紧密交互，并且可以对Table API中定义的表执行SQL查询。
   此外，SQL语言具有比较低的学习成本，能够让数据分析人员和开发人员快速上手。

Flink编程模型——程序结构

在Hadoop中，实现一个MapReduce应用程序需要编写Map和Reduce两部分；在Storm中，实现一个Topology需要编写Spout和Bolt两部分；同样，实现一个Flink应用程序也需要同样的逻辑。

一个Flink应用程序由3部分构成，或者说将Flink的操作算子可以分成3部分，分别为Source、Transformation和Sink，如图：

Flink编程模型

• Source：数据源部分。负责读取指定存储介质中的数据，转为分布式数据流或数据集，例如readTextFile()、socketTextStream()等算子。
• Transformation：数据转换部分。负责对一个或多个数据流或数据集进行各种转换操作，并产生一个或多个输出数据流或数据集，例如map()、flatMap()、keyBy()等算子。
• Sink：数据输出部分。负责将转换后的结果数据发送到HDFS、文本文件、MySQL、Elasticsearch等目的地，例如writeAsText()算子。

Flink编程模型-程序结构

第二章 Flink运行架构及原理

1、Flink运行时架构

1.1 Flink运行时架构-YARN架构

Flink有多种运行模式，可以运行在一台机器上，称为本地（单机）模式；也可以使用YARN或Mesos作为底层资源调度系统以分布式的方式在集群中运行，称为Flink On YARN模式（目前企业中使用最多的模式）；还可以使用Flink自带的资源调度系统，不依赖其他系统，称为Flink Standalone模式。

本地模式通常用于对应用程序的简单测试。

YARN集群总体上是经典的主/从（Master/Slave）架构，主要由ResourceManager、NodeManager、ApplicationMaster和Container等几个组件构成，YARN集群架构如图：

YARN架构

1. ResourceManager
   以后台进程的形式运行，负责对集群资源进行统一管理和任务调度。

2. NodeManager
   集群中每个节点上的资源和任务管理器，以后台进程的形式运行。它会定时向ResourceManager汇报本节点上的资源（内存、CPU）使用情况和各个Container的运行状态，同时会接收并处理来自ApplicationMaster的Container启动/停止等请求。

3. Task
   应用程序具体执行的任务。一个应用程序可能有多个任务，例如一个MapReduce程序可以有多个Map任务和多个Reduce任务。

4. Container
   YARN中资源分配的基本单位，封装了CPU和内存资源的一个容器，相当于一个Task运行环境的抽象。

5. ApplicationMaster
   应用程序管理者主要负责应用程序的管理，以后台进程的形式运行，为应用程序向ResourceManager申请资源（CPU、内存），并将资源分配给所管理的应用程序的Task。

YARN集群中应用程序的执行流程如图。

YARN集群应用程序的执行流程图

1. 客户端提交应用程序到ResourceManager。

2. ResourceManager分配用于运行ApplicationMaster的Container，然后与NodeManager通 信，要求它在该Container中启动ApplicationMaster。ApplicationMaster启动后，它将负责此应用程序的整个生命周期。

3. ApplicationMaster向ResourceManager注册（注册后可以通过ResourceManager查看应用程序的运行状态）并请求运行应用程序各个Task所需的Container（资源请求是对一些Container的请求）。如果符合条件，ResourceManager会分配给 ApplicationMaster所需的Container。

4. ApplicationMaster请求NodeManager使用这些Container来运行应用程序的相应Task（即将Task发布到指定的Container中运行）。

1.2 Flink运行时架构——Standalone架构

Flink Standalone模式为经典的主从（Master/Slave）架构，资源调度是Flink自己实现的。集群启动后，主节点上会启动一个JobManager进程，类似YARN集群的ResourceManager，因此主节点也称为JobManager节点；各个从节点上会启动一个TaskManager进程，类似YARN集群的NodeManager，因此从节点也称为TaskManager节点。

从Flink 1.6版本开始，将主节点上的进程名称改为了StandaloneSessionClusterEntrypoint，从节点的进程名称改为了TaskManagerRunner，在这里为了方便使用，仍然沿用之前版本的称呼，即JobManager和TaskManager。Flink Standalone模式的运行架构如图：

Flink StandAlong架构

Client接收到Flink应用程序后，将作业提交给JobManager。JobManager要做的第一件事就是分配Task（任务）所需的资源。完成资源分配后，Task将被JobManager提交给相应的TaskManager，TaskManager会启动线程开始执行。在执行过程中，TaskManager会持续向JobManager汇报状态信息，例如开始执行、进行中或完成等状态。作业执行完成后，结果将通过JobManager发送给Client。

1. Client
   Client是提交作业的客户端，虽然不是运行时和作业执行时的一部分，但它负责准备和提交作业到JobManager，它可以运行在任何机器上，只要与JobManager环境连通即可。提交完成后，Client可以断开连接，也可以保持连接来接收进度报告。

2. JobManager
   JobManager根据客户端提交的应用将应用分解为子任务，从资源管理器（YARN等）申请所需的计算资源，然后分发任务到TaskManager执行，并跟踪作业的执行状态等。

3. TaskManager
   TaskManager是Flink集群的工作进程。Task被调度到TaskManager上执行。TaskManager 相互通信，只为在后续的Task之间交换数据。

4. Task
   Flink中的每一个操作算子称为一个Task（任务），例如单词计数中使用的flatMap()算子、map()算子等。每个Task在一个JVM线程中执行。

5. Task Slot
   TaskManager为了控制执行的Task数量，将计算资源（内存）划分为多个Task Slot（任务槽），每个Task Slot代表TaskManager的一份固定内存资源，Task则在Task Slot中执行。例如，具有3个Task Slot的TaskManager会将其管理的内存资源分成3等份给每个Task Slot。
   每个Task Slot只对应一个执行线程。

1.3 Flink运行时架构——On YARN架构

Flink On YARN模式遵循YARN的官方规范，YARN只负责资源的管理和调度，运行哪种应用程序由用户自己实现，因此可能在YARN上同时运行MapReduce程序、Spark程序、Flink程序等。YARN很好地对每一个程序实现了资源的隔离，这使得Spark、MapReduce、Flink等可以运行于同一个集群中，共享集群存储资源与计算资源。

Flink On YARN模式的运行架构如图：

Flink On YARN架构

1. 当启动一个Client（客户端）会话时，Client首先会上传Flink应用程序JAR包和配置文件到HDFS。

2. Client向ResourceManager申请用于运行ApplicationMaster的Container。

3. ResourceManager分配用于运行ApplicationMaster的Container，然后与NodeManager通 信，要求它在该Container中启动ApplicationMaster（ApplicationMaster与Flink JobManager运行于同一Container中，这样ApplicationMaster就能知道Flink JobManager的地址）。ApplicationMaster启动后，它将负责此应用程序的整个生命周期。另外，ApplicationMaster还提供了Flink的WebUI服务。

4. ApplicationMaster向ResourceManager注册（注册后可以通过ResourceManager查看应用程序的运行状态）并请求运行Flink TaskManager所需的Container（资源请求是对一些Container的请求）。如果符合条件，ResourceManager会分配给 ApplicationMaster所需的Container。ApplicationMaster请求NodeManager使用这些Container来运行Flink TaskManager。各个NodeManager从HDFS中下载Flink JAR包和配置文件。至此，Flink相关任务就可以运行了。

此外，各个运行中的Flink TaskManager会通过RPC协议向ApplicationMaster汇报自己的状态和进度。

2、Flink任务调度原理

2.1 Flink任务调度原理——任务链

Flink中的每一个操作算子称为一个Task（任务），算子的每个具体实例则称为SubTask（子任务），SubTask是Flink中最小的处理单元，多个SubTask可能在不同的机器上执行。一个TaskManager进程包含一个或多个执行线程，用于执行SubTask。TaskManager中的一个Task Slot对应一个执行线程，一个执行线程可以执行一个或多个SubTask，如下图所示。

Task Slot对应一个执行线程

由于每个SubTask只能在一个线程中执行，为了能够减少线程间切换和缓冲的开销，在降低延迟的同时提高整体吞吐量，Flink可以将多个连续的SubTask链接成一个Task在一个线程中执行。这种将多个SubTask连在一起的方式称为任务链。下方右图中，一个Source类算子的SubTask和一个map()算子的SubTask连在了一起，组成了任务链。

SubTask任务链

2.2 Flink任务调度原理——并行度

Flink应用程序可以在分布式集群上并行运行，其中每个算子的各个并行实例会在单独的线程中独立运行，并且通常情况下会在不同的机器上运行。

为了充分利用计算资源，提高计算效率，可以增加算子的实例数（SubTask数量）。一个特定算子的SubTask数量称为该算子的并行度，且任意两个算子的并行度之间是独立的，不同算子可能拥有不同的并行度。例如，将Source算子、map()算子、keyby()/window()/apply()算子的并行度设置为2，Sink算子的并行度设置为1，运行效果如图：

Flink 任务调度-并行度

由于一个Task Slot对应一个执行线程，因此并行度为2的算子的SubTask将被分配到不同的Task Slot中执行。假设一个作业图（JobGraph）有A、B、C、D、E五个算子，其中A、B、D的并行度为4，C、E的并行度为2，该作业在TaskManager中的详细数据流程可能如图：

Flink任务调度-并行度

Flink中并行度的设置有4种级别：算子级别、执行环境（Execution Environment）级别、客户端（命令行）级别、系统级别。

1. 算子级别
   每个算子、Source和Sink都可以通过调用setParallelism()方法指定其并行度。例如以下代码设置flatMap()算子的并行度为2：

data.flatMap(_.split(" ")).setParallelism(2)

2. 执行环境级别
   调用执行环境对象的setParallelism()方法可以指定Flink应用程序中所有算子的默认并行度，代码如下：

val env=ExecutionEnvironment.getExecutionEnvironment
env.setParallelism(2)

3. 客户端（命令行）级别
   在向集群提交Flink应用程序时使用-p选项可以指定并行度。例如以下提交命令：

bin/flink run -p 2 WordCount.jar

4. 系统级别
   影响所有运行环境的系统级别的默认并行度可以在配置文件flink-conf.yaml中的parallelism.default属性中指定，默认为1。

4种并行度级别的作用顺序为：算子级别>执行环境级别>客户端级别>系统级别。

2.3 Flink任务调度原理——共享Task Slot

默认情况下，Flink允许SubTask之间共享Task Slot，即使它们是不同Task（算子）的SubTask，只要它们来自同一个作业（Job）即可。在没有共享Task Slot的情况下，简单的SubTask（source()、map()等）将会占用和复杂的SubTask（keyBy()、window()等）一样多的资源，通过共享Task Slot可以充分利用Task Slot的资源，同时确保繁重的SubTask在TaskManager之间公平地获取资源。例如，将算子并行度从2增加到6，并行效果如图:

共享Task Slot

最左侧的Task Slot负责作业的整个管道（Pipeline）。管道用于连接多个算子，将一个算子的执行结果输出给下一个算子。

2.4 Flink任务调度原理——数据流

一个Flink应用程序会被映射成逻辑数据流（Dataflow），而Dataflow都是以一个或多个Source开始、以一个或多个Sink结束的，且始终包括Source、Transformation、Sink三部分。Dataflow描述了数据如何在不同算子之间流动，将这些算子用带方向的直线连接起来会形成一个关于计算路径的有向无环图，称为DAG（Directed Acyclic Graph，有向无环图）或Dataflow图。各个算子的中间数据会被保存在内存中，如图：

Flink任务调度-数据流

假设一个Flink应用程序在读取数据后先对数据进行了map()操作，然后进行了keyBy()/window()/apply()操作，最后将计算结果输出到了指定的文件中，则该程序的Dataflow图如图：

Flink任务调度-数据流

假设该程序的Source、map()、keyBy()/window()/apply()算子的并行度为2，Sink算子的并行度为1，则该程序的逻辑数据流图、物理（并行）数据流图和Flink优化后的数据流图如图。

Flink数据流

Flink应用程序在执行时，为了降低线程开销，会将多个SubTask连接在一起组成任务链，在一个线程中运行。对于物理（并行）数据流来说，Flink执行时会对其进行优化，将Source[1]和map()[1]、Source[2]和map()[2]分别连接成一个任务，这是因为Source和map()之间采用了一对一的直连模式，而且没有任何的重分区，它们之间可以直接通过缓存进行数据传递，而不需要通过网络或序列化（如果不使用任务链，Source和map()可能在不同的机器上，它们之间的数据传递就需要通过网络）。这种优化在很大程度上提升了Flink的执行效率。

2.5 Flink任务调度原理——执行图

Flink应用程序执行时会根据数据流生成多种图，每种图对应了作业的不同阶段，根据不同图的生成顺序，主要分为4层：StreamGraph→JobGraph→ExecutionGraph→物理执行图，如图。

Flink任务调度-执行图

• StreamGraph：流图。使用DataStream API编写的应用程序生成的最初的图代表程序的拓扑结构，描述了程序的执行逻辑。StreamGraph在Flink客户端中生成，在客户端应用程序最后调用execute()方法时触发StreamGraph的构建。

• JobGraph：作业图。所有高级别API都需要转换为JobGraph。StreamGraph经过优化（例如任务链）后生成了JobGraph，以提高执行效率。StreamGraph和JobGraph都是在本地客户端生成的数据结构，而JobGraph需要被提交给JobManager进行解析。

• ExecutionGraph：执行图。JobManager对JobGraph进行解析后生成的并行化执行图是调度层最核心的数据结构。它包含对每个中间数据集或数据流、每个并行任务以及它们之间的通信的描述。

• 物理执行图：JobManager根据ExecutionGraph对作业进行调度后，在各个TaskManager上部署Task后形成的“图”。物理执行图并不是一个具体的数据结构，而是各个Task分布在不同的节点上所形成的物理上的关系表示。

2.6 Flink任务调度原理——执行计划

Flink的优化器会根据数据量或集群机器数等的不同自动地为程序选择执行策略。因此，准确地了解Flink如何执行编写的应用程序是很有必要的。

接下来我们对流处理单词计数例子的代码进行更改，从执行环境级别设置并行度 为2，代码如下：

val env=StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(2)
//将最后的触发任务执行代码env.execute("StreamWordCount")改为：
println(env.getExecutionPlan)//打印计划描述

直接在本地运行程序，将从控制台打印应用程序的逻辑执行计划对应的JSON描述，JSON字符串内容如下：

Flink任务调度-执行计划

Flink为执行计划提供了可视化工具，它可以把用JSON格式表示的作业执行计划以图的形式展现，并且其中包含完整的执行策略标注。

将上述JSON字符串粘贴到可视化工具网址（http://flink.apache.org/visualizer/）提供的文本框中，可将JSON字符串解析为可视化图，该可视化图对应的是StreamGraph，如图:

执行计划

将应用程序提交到Flink集群后，在Flink的WebUI中还可以看到另一张可视化图，即JobGraph，如图所示。此处的单词数据来源于HDFS存储系统，而不是本地文件。

执行计划

3、Flink数据分区

在Flink中，数据流或数据集被划分成多个独立的子集，这些子集分布到了不同的节点上，而每一个子集称为分区（Partition）。因此可以说，Flink中的数据流或数据集是由若干个分区组成的。数据流或数据集与分区的关系如图：

Flink数据分区

通过将每个记录分配给一个或多个分区来把数据流或数据集划分为多个分区。在运行期间，Task会消费数据流或数据集的分区。改变数据流或数据集分区方式的转换通常称为重分区。

3.1 Flink数据分区——分区数量

运行期间，每个数据记录将被分配给一个或多个分区，各个分区中的数据可以并行计算。数据是由上游算子的某个实例（SubTask）发往下游算子的一个或多个实例，而一个算子实例只负责计算一个分区的数据。因此，分区的数量是由下游算子的实例数量（并行度）决定的，发往下游算子的数据分区数量等于下游算子的实例数量。

如图所示，上游Source算子的并行度为1，下游map()算子的并行度为2，数据由Source发往map()，因此将分成两个分区，map()的两个实例各自执行一个分区的数据。

Flink数据分区-分区数量

数据分区的一个原则是使得分区的数量尽量等于集群节点CPU的核心数量，而前面提到过，算子的并行度应尽量等于集群节点CPU的核心数量，可见两者保持一致。

3.2 Flink数据分区——分区策略

Flink分区策略决定了一条数据如何发送给下游算子的不同实例，或者说如何在下游算子不同实例之间进行数据划分。程序运行时，系统会根据算子的语义和配置的并行度自动选择数据的分区策略。当然也可以在程序中显式指定分区策略。
Flink常见的分区策略如下。

1. 转发策略
   在上游算子实例和下游算子实例之间一对一地进行数据传输。这种策略不会产生重分区（改变数据流或数据集分区方式的转换通常称为重分区），且可以避免网络传输，以提高传输效率。假设上游算子A和下游算子B的并行度都为2，使用转发策略的效果如图所示。

转发策略

2. 广播策略
   上游算子实例的每个数据记录都会发往下游算子的所有实例。这种策略会把数据复制多份，向下游算子的每个实例发送一份，且涉及网络传输，代价较高。使用广播策略的效果如图：

广播策略

3. 键值策略
   根据数据记录中的键对数据进行重分区，键相同的数据记录一定会被发送给下游同一个算子实例，键不同的数据记录可能会被发送到下游不同的算子实例，也可能会被发送到下游同一个算子实例。这种策略要求数据记录的格式为(键,值)形式，如图：

键值策略

4. 随机策略
   将数据记录进行随机重分区，数据记录会被均匀分配到下游算子的每个实例。这种策略可以实现计算任务的负载均衡，如图：

随机策略

5. 全局策略
   将上游所有数据记录发送到下游第一个算子实例，如图：

全局策略

6. 自定义策略
   如果内置的分区策略不能满足当前需求，则可以在程序中自定义分区策略。

第三章 Flink安装及部署

1. Flink集群搭建

Flink可以在Linux、macOS和Windows上运行。前提条件是集群各节点提前安装JDK8以上版本，并配置好SSH免密登录，因为集群各节点之间需要相互通信，Flink主节点需要对其他节点进行远程管理和监控。

从Flink官网下载页面（https://flink.apache.org/downloads.html）下载二进制安装文件，并选择对应的Scala版本，此处选择Apache Flink 1.13.0 for Scala 2.11（Flink版本为1.13.0，使用的Scala版本为2.11）。

由于当前版本的Flink不包含Hadoop相关依赖库，如果需要结合Hadoop（例如读取HDFS中的数据），还需要下载预先捆绑的Hadoop JAR包，并将其放置在Flink安装目录的lib目录中。

此处选择Pre-bundled Hadoop 2.8.3（适用于Hadoop 2.8.3），如图:

Flink集群安装

接下来使用3个节点（主机名分别为centos01、centos02、centos03）讲解Flink各种运行模式的搭建。3个节点的主机名与IP的对应关系如表

Flink集群对应服务器

1.1 Flink本地模式

接下来讲解在CentOS 7操作系统中搭建Flink本地模式。

1. 上传解压安装包
   将下载的Flink安装包flink-1.13.0-bin-scala_2.11.tgz上传到centos01节点的/opt/softwares目录，然后进入该目录，执行以下命令将其解压到目录/opt/modules中。

$ tar -zxvf flink-1.13.0-bin-scala_2.11.tgz -C /opt/modules/

2. 启动Flink
   进入Flink安装目录，执行以下命令启动Flink：

$ bin/start-cluster.sh

启动后，使用jps命令查看Flink的JVM进程，命令如下：

$ jps
13309 StandaloneSessionClusterEntrypoint
13599 TaskManagerRunner

若出现上述进程，则代表启动成功。StandaloneSessionClusterEntrypoint为Flink主进程，即JobManager；TaskManagerRunner为Flink从进程，即TaskManager。

3. 查看WebUI
   在浏览器中访问服务器8081端口即可查看Flink的WebUI，此处访问地址http://192.168.170.133:8081/，如图:

Flink WebUI

从WebUI中可以看出，当前本地模式的Task Slot数量和TaskManager数量都为1（Task Slot数量默认为1）。

1.2 Flink集群搭建——Standalone模式

Flink Standalone模式的搭建需要在集群的每个节点都安装Flink，集群角色分配如表:

Flink Standalong模式

集群搭建的操作步骤如下：

1. 上传解压安装包
   将下载的Flink安装包flink-1.13.0-bin-scala_2.11.tgz上传到centos01节点的/opt/softwares目录，然后进入该目录，执行以下命令将其解压到目录/opt/modules中。

$ tar -zxvf flink-1.13.0-bin-scala_2.11.tgz -C /opt/modules/

2. 修改配置文件
   Flink的配置文件都存放于安装目录下的conf目录，进入该目录，执行以下操作。
   （1）修改flink-conf.yaml文件

$ vim conf/flink-conf.yaml

将文件中jobmanager.rpc.address属性的值改为centos01，命令如下：

jobmanager.rpc.address: centos01

上述配置表示指定集群主节点（JobManager）的主机名（或IP），此处为centos01。

（2）修改workers文件
workers文件必须包含所有需要启动的TaskManager节点的主机名，且每个主机名占一行。
执行以下命令修改workers文件：

$ vim conf/workers

改为以下内容：

centos02
centos03

上述配置表示将centos02和centos03节点设置为集群的从节点（TaskManager节点）。

3. 复制Flink安装文件到其他节点
   在centos01节点中进入/opt/modules/目录执行以下命令，将Flink安装文件复制到其他节点：

$ scp -r flink-1.13.0/ centos02:/opt/modules/
$ scp -r flink-1.13.0/ centos03:/opt/modules/

4. 启动Flink集群
   在centos01节点上进入Flink安装目录，执行以下命令启动Flink集群：

$ bin/start-cluster.sh

启动完毕后，分别在各节点执行jps命令，查看启动的Java进程。若各节点存在以下进程，则说明集群启动成功。

centos01节点：StandaloneSessionClusterEntrypoint
centos02节点：TaskManagerRunner
centos03节点：TaskManagerRunner

5. 查看WebUI
   集群启动后，在浏览器中访问JobManager节点的8081端口即可查看Flink的WebUI，此处访问地址http://192.168.170.133:8081/，如图:

Standalong模式的WebUI

从WebUI中可以看出，当前集群总的Task Slot数量（每个节点的Task Slot数量默认为1）和TaskManager数量都为2。

1.3 Flink集群搭建——On YARN模式

Flink On YARN模式的搭建比较简单，仅需要在YARN集群的一个节点上安装Flink即可，该节点可作为提交Flink应用程序到YARN集群的客户端。

若要在YARN上运行Flink应用，则需要注意以下几点：
1）Hadoop版本应在2.2以上。

2）必须事先确保环境变量文件中配置了HADOOP_CONF_DIR、YARN_CONF_DIR或者HADOOP_HOME，Flink客户端会通过该环境变量读取YARN和HDFS的配置信息，以便正确加载Hadoop配置以访问YARN，否则将启动失败。

3）需要下载预先捆绑的Hadoop JAR包，并将其放置在Flink安装目录的lib目录中，本例使用flink-shaded-hadoop-2-uber-2.8.3-10.0.jar。具体下载方式见3.1节的Flink集群搭建。

4）需要提前将HDFS和YARN集群启动。

本例使用的Hadoop集群各节点的角色分配如表：

Flink On YARN

在Flink On YARN模式中，根据作业的运行方式不同，又分为两种模式：Flink YARN Session模式和Flink Single Job（独立作业）模式。

Flink YARN Session模式需要先在YARN中启动一个长时间运行的Flink集群，也称为Flink YARN Session集群，该集群会常驻在YARN集群中，除非手动停止。客户端向Flink YARN Session集群中提交作业时，相当于连接到一个预先存在的、长期运行的Flink集群，该集群可以接受多个作业提交。即使所有作业完成后，集群（和JobManager）仍将继续运行直到手动停止。该模式下，Flink会向YARN一次性申请足够多的资源，资源永久保持不变，如果资源被占满，则下一个作业无法提交，只能等其中一个作业执行完成后释放资源，如图：

Flink YARN Session

拥有一个预先存在的集群可以节省大量时间申请资源和启动TaskManager。作业可以使用现有资源快速执行计算是非常重要的。

Flink Single Job模式不需要提前启动Flink YARN Session集群，直接在YARN上提交Flink作业即可。每一个作业会根据自身情况向YARN申请资源，不会影响其他作业运行，除非整个YARN集群已无任何资源。并且每个作业都有自己的JobManager和TaskManager，相当于为每个作业提供了一个集群环境，当作业结束后，对应的组件也会同时释放。该模式不会额外占用资源，使资源利用率达到最大，在生产环境中推荐使用这种模式，如图：

Flink Single Job

Flink Single Job模式适合长期运行、具有高稳定性要求且对较长的启动时间不敏感的大型作业。

1. Flink YARN Session模式操作
   （1）启动Flink YARN Session集群
   在启动HDFS和YARN集群后，在YARN集群主节点（此处为centos01节点）安装好Flink，进入Flink主目录执行以下命令，即可启动Flink YARN Session集群：

$ bin/yarn-session.sh -jm 1024 -tm 2048

上述命令中的参数-jm表示指定JobManager容器的内存大小（单位为MB），参数-tm表示指定TaskManager容器的内存大小（单位为MB）。

启动完毕后，会在启动节点（此处为centos01节点）产生一个名为FlinkYarnSessionCli的进程，该进程是Flink客户端进程；在其中一个NodeManager节点产生一个名为YarnSessionClusterEntrypoint的进程，该进程是Flink JobManager进程。而Flink TaskManager进程不会启动，在后续向集群提交作业时才会启动。例如，启动完毕后查看centos01节点的进程可能如下：

$ jps
7232 NodeManager
6626 NameNode
14422 YarnSessionClusterEntrypoint
14249 FlinkYarnSessionCli
6956 SecondaryNameNode
7116 ResourceManager
17612 Jps
6750 DataNode

此时可以在浏览器访问YARN ResourceManager节点的8088端口，此处地址为http://192.168.170.133:8088/，在YARN的WebUI中可以查看当前Flink应用程序（Flink YARN Session集群）的运行状态，如图

Flink YARN Session集群

从图中可以看出，一个Flink YARN Session集群实际上就是一个长时间在YARN中运行的应用程序（Application），后面的Flink作业也会提交到该应用程序中。

（2）提交Flink作业
接下来向Flink YARN Session集群提交Flink自带的单词计数程序。
首先在HDFS中准备/input/word.txt文件，内容如下：

hello hadoop
hello java
hello scala
java

然后在Flink客户端（centos01节点）中执行以下命令，提交单词计数程序到Flink YARN Session集群：

$ bin/flink run ./examples/batch/WordCount.jar \
-input hdfs://centos01:9000/input/word.txt \
-output hdfs://centos01:9000/result.txt

上述命令通过参数-input指定输入数据目录，-output指定输出数据目录。
在执行过程中，查看Flink YARN Session集群的WebUI，如图:

Flink客户端执行单词计数程序

当作业执行完毕后，查看HDFS/result.txt文件中的结果，如图：

Flink客户端单词计数程序执行结果

（3）分离模式
如果希望将启动的Flink YARN Session集群在后台独立运行，与Flink客户端进程脱离关系，可以在启动时添加-d或--detached参数，表示以分离模式运行作业，即Flink客户端在启动Flink YARN Session集群后，就不再属于YARN集群的一部分。例如以下代码：

$ bin/yarn-session.sh -jm 1024 -tm 2048 -d

（4）进程绑定
与分离模式相反，当使用分离模式启动Flink YARN Session集群后，如果需要再次将Flink客户端与Flink YARN Session集群绑定，则使用-id或--applicationId参数指定Flink YARN Session集群在YARN中对应的applicationId即可，命令格式如下：

$ bin/yarn-session.sh –id [applicationId]

例如，将Flink客户端（执行绑定命令的本地客户端）与applicationId为application_ 1593999118637_0009的Flink YARN Session集群绑定，命令如下：

$ bin/yarn-session.sh -id application_1593999118637_0009

执行上述命令后，在Flink客户端会产生一个名为FlinkYarnSessionCli的客户端进程。此时就可以在Flink客户端对Flink YARN Session集群进行操作，包括执行停止命令等。例如执行Ctrl+C命令或输入stop命令即可停止Flink YARN Session集群。

2. Flink Single Job模式操作

Flink Single Job模式可以将单个作业直接提交到YARN中，每次提交的Flink作业都是一个独立的YARN应用程序，应用程序运行完毕后释放资源，这种模式适合批处理应用。

例如，在Flink客户端（centos01节点）中执行以下命令，以Flink Single Job模式提交单词计数程序到YARN集群：

$ bin/flink run -m yarn-cluster examples/batch/WordCount.jar \
-input hdfs://centos01:9000/input/word.txt \
-output hdfs://centos01:9000/result.txt

上述命令通过参数-m指定使用YARN集群（即以Flink Single Job模式提交），-input指定输入数据目录，-output指定输出数据目录。
提交完毕后，可以在浏览器访问YARN ResourceManager节点的8088端口，此处地址为http://192.168.170.133:8088/，在YARN的WebUI中可以查看当前Flink应用程序的运行状态，如图:

image.png

2. Flink HA模式

3. Flink命令行界面

4. Flink应用提交

5. Flink Shell的使用

第四章 Flink DataStream API

01 基本概念

DataStream API的名称来自一个特殊的DataStream类，该类用于表示Flink程序中的数据集合。你可以将它视为包含重复项的不可变数据集合。这些数据可以是有限的，也可以是无限的，用于处理这些数据的API是相同的。

Flink中使用DataSet和DataStream表示数据的基本抽象，可以将它们视为包含特定类型的元素集合，类似于常规Java集合。但不同的是，集合数据不可变，集合一旦被创建，就不能添加或删除元素。对于DataSet，数据是有限的，而对于DataStream，元素的数量可以是无限的。

DataSet和DataStream数据集都是分布式数据集，分布式数据集是指：一个数据集存储在不同的服务器节点上，每个节点存储数据集的一部分。例如，将数据集(hello,world,scala,spark,love,spark,happy)存储在3个节点上，节点一存储(hello,world)，节点二存储(scala,spark,love)，节点三存储(spark,happy)，这样对3个节点的数据可以并行计算，并且3个节点的数据共同组成了一个DataSet/DataStream，如图：

DataSet集合

分布式数据集类似于HDFS中的文件分块，不同的块存储在不同的节点上；而并行计算类似于使用MapReduce读取HDFS中的数据并进行Map和Reduce操作。Flink包含这两种功能，并且计算更加灵活。

DataSet/DataStream数据集的全部或部分可以缓存在内存中，并且可以在多次计算时重用。数据也可以持久化到磁盘，具有高效的容错能力。

DataSet的主要特征如下（DataStream同样拥有）：

• 数据是不可变的，但可以将DataSet转换成新的DataSet进行操作。

• 数据是可分区的。DataSet由很多分区组成，每个分区对应一个Task任务来执行。

• 对DataSet进行操作，相当于对每个分区进行操作。

• DataSet拥有一系列对分区进行计算的函数，称为算子（关于算子将在4.6节详细讲解）。

• DataSet之间存在依赖关系，可以实现管道化，避免了中间数据的存储。

• 在编程时，可以把DataSet/DataStream看作一个数据操作的基本单位，而不必关心数据的分布式特性，Flink会自动将其中的数据分发到集群的各个节点。Flink中对数据的操作主要是对DataSet/DataStream的操作（创建、转换、求值等）。

02 执行模式

DataStream API 支持不同的运行时执行模式，可以根据用例的要求和作业的特征从中进行选择。DataStream API比较“经典”的执行行为称为“流”执行模式，主要用于需要连续增量处理并无限期保持在线的无限作业。

此外，还有一种“批”处理执行模式。该模式以一种类似于MapReduce等批处理框架的方式执行作业，主要用于具有已知固定输入并且不会连续运行的有界作业。

当需要为最终使用无界源运行的代码编写测试时，可以使用“流”模式运行有界作业。在测试情况下使用有界源会更自然。

不管配置的执行模式如何，在有界输入上执行的DataStream应用程序都会产生相同的最终结果。以流模式执行的作业可能会产生增量更新，而批作业最终只会产生一个最终结果。

通过启用批执行，允许Flink应用额外的优化。

执行模式可以通过execution.runtime-mode属性来配置。其有3种可能的值：

• STREAMING：典型的DataStream执行模式（默认）。
• BATCH：在DataStream API上以批处理方式执行。
• AUTOMATIC：让系统根据数据源的有界性来决定。

也可以通过使用bin/flink run命令行参数配置，或在Flink应用程序中创建StreamExecutionEnvironment对象时以编程方式指定。

通过命令行配置执行模式，代码如下：

$ bin/flink run -Dexecution.runtime-mode=BATCH examples/streaming/WordCount.jar

上述代码向Flink集群中提交了单词计数程序，并指定使用批执行模式。
在Flink应用程序中通过代码配置执行模式，代码如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment. getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

03 作业流程

Flink的作业执行流程如图:

作业流程

Flink JobManager是Flink集群的主节点。它包含3个不同的组件：Flink Resource Manager、Dispatcher、运行每个Flink Job的JobMaster。

在一个作业提交前，JobManager和TaskManager等进程需要先被启动。可以在Flink安装目录中执行bin/start-cluster.sh命令来启动这些进程。JobManager和TaskManager被启动后，TaskManager需要将自己注册给JobManager中的ResourceManager（资源注册）。

Flink作业的具体执行流程如下：

1. 用户编写应用程序代码，并通过Flink客户端提交作业。程序一般为Java或Scala语言，调用Flink API构建逻辑数据流图，然后转为作业图JobGraph，并附加到StreamExecutionEnvironment中。代码和相关配置文件被编译打包，被提交到JobManager的Dispatcher，形成一个应用作业。
2. Dispatcher（JobManager的一个组件）接收到这个作业，启动JobManager，JobManager负责本次作业的各项协调工作。
3. 接下来JobManager向ResourceManager申请本次作业所需的资源。
4. JobManager将用户作业中的作业图JobGraph转化为并行化的物理执行图，对作业并行处理并将其子任务分发部署到多个TaskManager上执行。每个作业的并行子任务将在Task Slot中执行。至此，一个Flink作业就开始执行了。
5. TaskManager在执行计算任务的过程中可能会与其他TaskManager交换数据，会使用相应的数据交换策略。同时，TaskManager也会将一些任务状态信息反馈给JobManager，这些信息包括任务启动、运行或终止的状态、快照的元数据等。

04 程序结构

Flink DataStream程序都包含相同的基本部分：

1. 获取执行环境。
2. 加载/创建初始数据。
3. 对初始数据进行转换。
4. 指定计算结果的输出位置。
5. 触发程序执行。

StreamExecutionEnvironment是所有Flink流程序的基础。我们可以在StreamExecutionEnvironment上使用getExecutionEnvironment()创建一个执行环境。该方法将根据上下文自动获取当前正确的执行环境。如果是在IDE中执行程序或作为常规Java程序执行，它将创建一个本地环境，程序将在本地机器上执行。如果将程序打包成一个JAR文件，并通过集群的命令行执行它，那么Flink集群管理器将执行程序的main方法，而getExecutionEnvironment()将返回一个集群执行环境。

有多种方法可以为执行环境指定数据源，例如从CSV文件中逐行读取或从其他数据源中读取。按行读取文本文件中的数据，代码如下：

val env = StreamExecutionEnvironment.getExecutionEnvironment()
val text: DataStream[String] = env.readTextFile("file:///path/to/file")

上述代码将得到一个数据流，接下来可以在该数据流上应用转换算子来创建新的派生数据流。例如，将数据流中的每个元素转为整数，使用map转换代码如下：

val input: DataStream [String] = ...
val mapped = input.map { x => x.toInt }

上述代码通过将原始集合中的每个字符串转换为整数来创建一个新的数据流。
一旦有了包含最终结果的数据流，就可以通过创建接收器将流数据写入外部系统。创建接收器的示例方法代码如下：

//将数据流以字节数组的形式写入Socket
dataStream.writeToSocket()
//将数据流写入标准输出流(stdout)，数据流的每个元素将以toString的方式转为字符串
dataStream.print()

完整程序写完后，最后需要通过在StreamExecutionEnvironment上调用execute()来触发程序执行。根据执行环境的类型（本地或集群），执行将在本地计算机上触发或提交程序以在集群上执行。

所有Flink程序都是延迟（惰性）执行的：执行程序的main()方法时，不会直接进行数据加载和转换。而是将每个操作添加到数据流图。当在执行环境中调用execute()显式触发执行时才会执行这些操作。

05 Source数据源

5.1 Source数据源——基本数据源

DataStream API中直接提供了对一些基本数据源的支持，例如文件系统、Socket连接等；也提供了非常丰富的高级数据源连接器（Connector），例如Kafka Connector、Elasticsearch Connector等。用户也可以实现自定义Connector数据源，以便使Flink能够与其他外部系统进行数据交互。

1. 文件数据源
   Flink可以将文件内容读取到系统中，并转换成分布式数据集DataStream进行处理。
   使用readTextFile(path)方法可以逐行读取文本文件内容，并作为字符串返回，代码如下：

//第一步：创建流处理的执行环境
val senv=StreamExecutionEnvironment.getExecutionEnvironment
//第二步：读取流数据，创建DataStream
val data:DataStream[String]=senv
    .readTextFile("hdfs://centos01:9000/input/words.txt")

2. Socket数据源
   通过监听Socket端口接收数据创建DataStream。例如以下代码从本地的9999端口接收数据：

//第一步：创建流处理的执行环境
val senv=StreamExecutionEnvironment.getExecutionEnvironment
//第二步：读取流数据，创建DataStream
val data:DataStream[String]=senv.socketTextStream("localhost",9999)

3. 集合数据源
   从java.util.collection集合创建DataStream。集合中的所有元素必须是相同类型的，例如以下代码：

//第一步：创建流处理的执行环境
val senv=StreamExecutionEnvironment.getExecutionEnvironment
//第二步：读取流数据，创建DataStream
val data:DataStream[String]=senv.fromCollection(
   List("hello","flink","scala")
)

当然，也可以从迭代器中创建DataStream，例如以下代码：

//第一步：创建流处理的执行环境
val senv=StreamExecutionEnvironment.getExecutionEnvironment
//第二步：读取流数据，创建DataStream
val it = Iterator("hello","flink","scala")
val data:DataStream[String]=senv.fromCollection(it)

还可以直接从元素集合中创建DataStream，例如以下代码：

//第一步：创建流处理的执行环境
val senv=StreamExecutionEnvironment.getExecutionEnvironment
//第二步：读取流数据，创建DataStream
val data:DataStream[String]=senv.fromElements("hello","flink","scala")

5.2 Source数据源——高级数据源

Flink可以从Kafka、Flume、Kinesis等数据源读取数据，使用时需要引入第三方依赖库。例如，在Maven工程中引入Flink针对Kafka的API依赖库，代码如下：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka_2.11</artifactId>
    <version>1.13.0</version>
</dependency>

然后使用addSource()方法接入Kafka数据源，代码示例如下：

val senv = StreamExecutionEnvironment.getExecutionEnvironment()
val myConsumer = new FlinkKafkaConsumer08[String](...)
val stream = senv.addSource(myConsumer)

5.3 Source数据源——自定义数据源

在Flink中，用户也可以自定义数据源，以满足不同数据源的接入需求。自定义数据源有3种方式：
1）实现SourceFunction接口定义非并行数据源（单线程）。SourceFunction是Flink中所有流数据源的基本接口。
2）实现ParallelSourceFunction接口定义并行数据源。
3）继承RichParallelSourceFunction抽象类定义并行数据源。该类已经实现了ParallelSourceFunction接口，是实现并行数据源的基类，在执行时，Flink Runtime将执行与该类源代码配置的并行度一样多的并行实例。
4）继承RichSourceFunction抽象类定义并行数据源。该类是实现并行数据源的基类，该数据源可以通过父类AbstractRichFunction的getRuntimeContext()方法访问上下文信息，通过父类AbstractRichFunction的open()和close()方法访问生命周期信息。
数据源定义好后，可以使用StreamExecutionEnvironment.addSource(sourceFunction)将数据源附加到程序中。这样就可以将外部数据转换为DataStream。

例如，自定义MySQL数据源，读取MySQL中的表数据，实现步骤如下。
（1）引入数据库驱动
在Maven工程中引入MySQL数据库连接驱动的依赖库，代码如下：

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.49</version>
</dependency>

（2）创建表
在MySQL数据库中创建一张student表并添加测试数据，如图:

Flink数据源-mysql数据

（3）定义样例类
定义样例类Student用于存储数据，代码如下：

package flink.demo
object Domain {
    case class Student(id: Int, name: String, age: Int)
}

（4）创建JDBC工具类
创建一个JDBC工具类，用于获得MySQL数据库连接，代码如下：

import java.sql.DriverManager
import java.sql.Connection

/**
  * JDBC工具类
  */
object JDBCUtils {
    //数据库驱动类
    private val driver = "com.mysql.jdbc.Driver"
    //数据库连接地址
    private val url = "jdbc:mysql://localhost:3306/student_db"
    //数据库账号
    private val username = "root"
    //数据库密码
    private val password = "123456"

    /**
      * 获得数据库连接
      */
    def getConnection(): Connection = {
        Class.forName(driver)//加载驱动
        val conn = DriverManager.getConnection(url, username, password)
        conn
    }
}

（5）创建自定义数据源类
创建自定义MySQL数据源类MySQLSource，继承RichSourceFunction类，并重写open()、run()、cancel()方法，代码如下：

import java.sql.{Connection, PreparedStatement}
import flink.demo.Domain.Student
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.configuration.Configuration

/**
  * 自定义MySQL数据源
  */
class MySQLSource extends RichSourceFunction[Student] {

    var conn: Connection = _//数据库连接对象
    var ps: PreparedStatement = _//SQL命令执行对象
    var isRunning=true//是否运行（是否持续从数据源读取数据）

    /**
      * 初始化方法
      * @param parameters 存储键/值对的轻量级配置对象
      */
    override def open(parameters: Configuration): Unit = {
        //获得数据库连接
        conn = JDBCUtils.getConnection
        //获得命令执行对象
        ps = conn.prepareStatement("select * from student")
    }

    /**
      * 当开始从数据源读取元素时，该方法将被调用
      * @param ctx 用于从数据源发射元素
      */
    override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {
        //执行查询
        val rs = ps.executeQuery()
        //循环读取集合中的数据并发射出去
        while (isRunning&&rs.next()) {
            val student = Student(
                rs.getInt("id"),
                rs.getString("name"),
                rs.getInt("age")
            )
            //从数据源收集一个元素数据并发射出去，而不附加时间戳（默认方式）
            ctx.collect(student)
        }
    }

    /**
      * 取消数据源读取
     */
    override def cancel(): Unit = {
        this.isRunning=false
    }
}

（6）测试程序
创建测试类StreamTest，从自定义数据源中读取流数据，打印到控制台，代码如下：

import org.apache.flink.streaming.api.scala.{DataStream, _}

/**
  * 测试类
  */
object StreamTest {
    def main(args: Array[String]): Unit = {
        //创建流处理执行环境
        val senv=StreamExecutionEnvironment.getExecutionEnvironment
         //从自定义数据源中读取数据，创建DataStream
        val dataStream: DataStream[Domain.Student] = senv.addSource(new MySQLSource)
        //打印流数据到控制台
        dataStream.print()
        //触发任务执行，指定作业名称
        senv.execute("StreamMySQLSource")
    }
}

直接在IDEA中运行上述测试类，控制台输出结果如下：
3> Student(2,李四,22)
2> Student(1,张三,19)
4> Student(3,王五,20)
结果前面的数字表示执行线程的编号。

06 Transformation数据转换

在Flink中，Transformation（转换）算子就是将一个或多个DataStream转换为新的DataStream，可以将多个转换组合成复杂的数据流（Dataflow）拓扑。
常用的Transformation算子介绍如下。

1. map(func)
   map()算子接收一个函数作为参数，并把这个函数应用于DataStream的每个元素，最后将函数的返回结果作为结果DataStream中对应元素的值，即将DataStream的每个元素转换成新的元素。

如以下代码所示，对dataStream1应用map()算子，将dataStream1中的每个元素加一并返回一个名为dataStream2的新DataStream：

val dataStream1=senv.fromCollection(List(1,2,3,4,5,6))
val dataStream2=dataStream1.map(x=>x+1)

Flink map算子

2. flatMap(func)
   与map()算子类似，但是每个传入该函数func的DataStream元素会返回0到多个元素，最终会将返回的所有元素合并到一个DataStream。

例如以下代码将集合List转为dataStream1，然后调用dataStream1的flatMap()算子将dataStream1的每个元素按照空格分割成多个元素，最终合并所有元素到一个新的DataStream。

//创建DataStream
val dataStream1=senv.fromCollection(
    List("hadoop hello scala","flink hello")
)
//调用flatMap()算子进行运算
val dataStream2=dataStream1.flatMap(_.split(" "))
//打印结果到控制台
dataStream2.print()
//触发任务执行，指定作业名称
senv.execute("MyJob")

Flink的flatMap算子

3. filter(func)
   通过函数func对源DataStream的每个元素进行过滤，并返回一个新的DataStream。

例如以下代码，过滤出dataStream1中大于3的所有元素，并输出结果。

val dataStream1=senv.fromCollection(List(1,2,3,4,5,6))
val dataStream2=dataStream1.filter(_>3)
dataStream2.print()

控制台输出结果如下：

1> 4
2> 5
3> 6

4. keyBy()
   keyBy()算子主要作用于元素类型是元组或数组的DataStream上。使用该算子可以将DataStream中的元素按照指定的key（指定的字段）进行分组，具有相同key的元素将进入同一个分区中（不进行聚合），并且不改变原来元素的数据结构。例如，根据元素的形状对元素进行分组，相同形状的元素将被分配到一起，可被后续算子统一处理，如图:

   
   
   Flink keyBy算子

假设有两个同学zhangsan和lisi，zhangsan的语文和数学成绩分别为98、78，lisi的语文和数学成绩分别为88、79。将数据集的姓名作为key进行keyBy()操作，代码如下：

val dataStream=senv.fromCollection(
    List(("zhangsan",98),("zhangsan",78),("lisi",88),("lisi",79))
)

//使用数字位置指定，按照第一个字段分组

val keyedStream: KeyedStream[(String, Int), Tuple]=dataStream.keyBy(_._1)
keyedStream.print()

控制台输出结果如下：

4> (lisi,88)
4> (lisi,79)
3> (zhangsan,98)
3> (zhangsan,78)

从上述输出结果可以看出，
同一组的元素被同一个线程执行。运行过程如图。

Flink keyBy算子

keyBy()算子的执行对象是DataStream，执行结果则是KeyedStream。KeyedStream实际上是一种特殊的DataStream，因为其继承了DataStream。KeyedStream用来表示根据指定的key进行分组的数据流。

5. reduce()

reduce()算子主要作用于KeyedStream上，对KeyedStream数据流进行滚动聚合，即将当前元素与上一个聚合值进行合并，并且发射出新值。该算子的原理与MapReduce中的Reduce类似，聚合前后的元素类型保持一致，如图所示。

Flink reduce算子

reduce()算子始终以滚动的方式将两个元素合并为一个元素，最终将一组元素合并为单个元素。reduce()算子可以用于整个数据集，也可以用于分组的数据集。该算子的执行效率比较高，因为它允许系统使用更有效的执行策略。

继续对前面两个同学zhangsan和lisi的成绩进行reduce()操作，求出每个同学的总成绩，代码如下：

val reducedDataStream: DataStream[(String, Int)] = keyedStream
  .reduce((t1, t2) => {
//聚合规则：将每一组的第二个字段进行累加，第一个字段保持不变。
//注意聚合后数据类型与聚合前保持一致(String, Int)
      (t1._1, t1._2 + t2._2)
  })
reducedDataStream.print()

6. Aggregation

除了reduce()算子外，其他常用的聚合算子有sum()、max()、min()等，这些聚合算子统称为Aggregation。Aggregation算子作用于KeyedStream上，并且进行滚动聚合。与keyBy()算子类似，可以使用数字或字段名称指定需要聚合的字段。例如以下代码：

keyedStream.sum(0);//对第一个字段进行求和
keyedStream.sum("key");//对字段key进行求和
keyedStream.min(0);
keyedStream.min("key");
keyedStream.max(0);
keyedStream.max("key");
keyedStream.minBy(0);
keyedStream.minBy("key");
keyedStream.maxBy(0);
keyedStream.maxBy("key");

keyBy()算子会将DataStream转换为KeyedStream，而Aggregation算子会将KeyedStream转换为DataStream。相关类型转换如图:

DataStream和KeyedStream转换

7. union()

union()算子用于将两个或多个数据流进行合并，创建一个包含所有数据流所有元素的新流（不会去除重复元素）。如果将一个数据流与它本身合并，在结果流中，每个元素会出现两次。使用union()算子合并数据流，如图：

Flink union算子

union()算子执行过程中，多条数据流中的元素会以先进先出的方式合并，无法保证顺序，每个输入的元素都会被发往下游算子。使用union()算子对两个数据流进行合并，代码如下：

//创建数据流一
val dataStream1 = senv.fromElements(
    (0, 0, 0), (1, 1, 1), (2, 2, 2)
)
//创建数据流二
val dataStream2 = senv.fromElements(
    (3, 3, 3), (4, 4, 4), (5, 5, 5)
)
//合并两个数据流
val unionDataStream=dataStream1.union(dataStream2)
unionDataStream.print()

07 Sink数据输出

Flink可以使用DataStream API将数据流输出到文件、Socket、外部系统等。Flink自带了各种内置的输出格式，说明如下。

• writeAsText()：将元素转为String类型按行写入外部输出。String类型是通过调用每个元素的toString()方法获得的。

• writeToSocket()：将元素写入Socket。

• writeAsCsv()：将元组写入以逗号分隔的文本文件。行和字段分隔符是可配置的。每个字段的值来自对象的toString()方法。

• addSink()：调用自定义接收函数。Flink可以与其他系统（如Apache Kafka）的连接器集成在一起，这些系统已经实现了自定义接收函数。

08 分区策略

数据在算子之间流动需要依靠分区策略（分区器），Flink目前内置了8种已实现的分区策略和1种自定义分区策略。已实现的分区策略对应的API为：

BinaryHashPartitioner
BroadcastPartitioner
ForwardPartitioner
GlobalPartitioner
KeyGroupStreamPartitioner
RebalancePartitioner
RescalePartitioner
ShufflePartitioner

自定义分区策略的API为CustomPartitionerWrapper。

1. BinaryHashPartitioner

该分区策略位于Blink的Table API的org.apache.flink.table.runtime.partitioner包中，是一种针对BinaryRowData的哈希分区器。BinaryRowData是RowData的实现，可以显著减少Java对象的序列化/反序列化。RowData用于表示结构化数据类型，运行时通过Table API或SQL管道传递的所有顶级记录都是RowData的实例。关于BinaryHashPartitioner，此处不做过多讲解。

2. BroadcastPartitioner

广播分区策略将上游数据记录输出到下游算子的每个并行实例中，即下游每个分区都会有上游的所有数据。使用DataStream的broadcast()方法即可设置该DataStream向下游发送数据时使用广播分区策略，Java代码如下：

dataStream.broadcast()

广播分区策略数据流图如下。

Flink广播分区策略

3. ForwardPartitioner

转发分区策略只将元素转发给本地运行的下游算子的实例，即将元素发送到与当前算子实例在同一个TaskManager的下游算子实例，而不需要进行网络传输。要求上下游算子并行度一样，这样上下游算子可以同属一个子任务。

使用DataStream的forward()方法即可设置该DataStream向下游发送数据时使用转发分区策略，Java代码如下：

dataStream.forward()

转发分区策略数据流图如下。

4. GlobalPartitioner

全局分区策略将上游所有元素发送到下游子任务编号等于0的分区算子实例上（下游第一个实例）。

使用DataStream的global()方法即可设置该DataStream向下游发送数据时使用全局分区策略，Java代码如下：

dataStream.global()

全局分区策略数据流图如下。

Flink全局分区

5. KeyGroupStreamPartitioner

Key分区策略根据元素Key的Hash值输出到下游算子指定的实例。keyBy()算子底层正是使用的该分区策略，底层最终会调用KeyGroupStreamPartitioner的selectChannel()方法，计算每个Key对应的通道索引（通道编号，可理解为分区编号），根据通道索引将Key发送到下游相应的分区中。

6. RebalancePartitioner

平衡分区策略使用循环遍历下游分区的方式，将上游元素均匀分配给下游算子的每个实例。每个下游算子的实例都具有相等的负载。当数据流中的元素存在数据倾斜时，使用该策略对性能有很大的提升。
使用DataStream的rebalance()方法即可设置该DataStream向下游发送数据时使用平衡分区策略，Java代码如下：

dataStream.rebalance()

平衡分区策略数据流图如下:

Flink平衡分区

7. RescalePartitioner

重新调节分区策略基于上下游算子的并行度，将元素以循环的方式输出到下游算子的每个实例。类似于平衡分区策略，但又与平衡分区策略不同。

上游算子将元素发送到下游哪一个算子实例，取决于上游和下游算子的并行度。例如，如果上游算子的并行度为2，而下游算子的并行度为4，那么一个上游算子实例将把元素均匀分配给两个下游算子实例，而另一个上游算子实例将把元素均匀分配给另外两个下游算子实例。相反，如果下游算子的并行度为2，而上游算子的并行度为4，那么两个上游算子实例将分配给一个下游算子实例，而另外两个上游算子实例将分配给另一个下游算子实例。

假设上游算子并行度为2，分区编号为A和B，下游算子并行度为4，分区编号为1、2、3、4，那么A将把数据循环发送给1和2，B则把数据循环发送给3和4。假设上游算子并行度为4，编号为A、B、C、D，下游算子并行度为2，编号为1、2，那么A和B把数据发送给1，C和D则把数据发送给2。
使用DataStream的rescale()方法即可设置该DataStream向下游发送数据时使用重新调节分区策略，Java代码如下：

dataStream.rescale()

重新调节分区策略数据流图如下:

重新调节分区

如果想将元素均匀地输出到下游算子的每个实例，以实现负载均衡，同时又不希望使用平衡分区策略的全局负载均衡，则可以使用重新调节分区策略。该策略会尽可能避免数据在网络间传输，而能否避免还取决于TaskManager的Task Slot数量、上下游算子的并行度等。

8. ShufflePartitioner

随机分区策略将上游算子元素输出到下游算子的随机实例中。元素会被均匀分配到下游算子的每个实例。这种策略可以实现计算任务的负载均衡。
使用DataStream的shuffle()方法即可设置该DataStream向下游发送数据时使用随机分区策略，Java代码如下：

dataStream.shuffle()

09 窗口计算

10 水印

11 状态管理

12 容错机制