Flink快速入门

Flink快速入门 Flink快速入门 Flink概念

Flink是什么？

官网介绍：Apache Flink® — Stateful Computations over Data Streams

翻译：Apache Flink ——数据流上的有状态计算

Flink定义：

Apache Flink 是一个框架和分布式处理引擎，用于在无边界和有边界数据流上进行有状态的计算。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。

数据可以被作为 无界 或者 有界 流来处理。

1.无界流：有定义流的开始，但没有定义流的结束。它们会无休止地产生数据。无界流的数据必须持续处理，即数据被摄取后需要立刻处理。我们不能等到所有数据都到达再处理，因为输入是无限的，在任何时候输入都不会完成。处理无界数据通常要求以特定顺序摄取事件，例如事件发生的顺序，以便能够推断结果的完整性。

2.有界流： 有定义流的开始，也有定义流的结束。有界流可以在摄取所有数据后再进行计算。有界流所有数据可以被排序，所以并不需要有序摄取。有界流处理通常被称为批处理。

Flink组件栈

从下至上：

1、部署：Flink 支持本地运行（IDE 中直接运行程序）、能在独立集群（Standalone 模式）或者在被 YARN、 Mesos、K8s 管理的集群上运行，也能部署在云上。

2、运行：Flink 的核心是分布式流式数据引擎，意味着数据以一次一个事件的形式被处理。

3、API：DataStream、DataSet、Table、SQL API。

4、扩展库：Flink 还包括用于 CEP（复杂事件处理）、机器学习、图形处理等场景。

Flink名词介绍

Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。

Flink整个系统包含三个部分：

•Client：

Flink Client主要给用户提供向Flink系统提交用户任务（流式作业）的能力。

•TaskManager：

Flink系统的业务执行节点，执行具体的用户任务。TaskManager可以有多个，各个TaskManager都平等。

•JobManager：

Flink系统的管理节点，管理所有的TaskManager，并决策用户任务在哪些Taskmanager执行。JobManager在HA模式下可以有多个，但只有一个主JobManager。

Flink系统提供的关键能力：

•低时延提供ms级时延的处理能力。

•Exactly Once提供异步快照机制，保证所有数据真正只处理一次。

•HAJobManager支持主备模式，保证无单点故障。

•水平扩展能力TaskManager支持手动水平扩展。

Flink最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。

Flink应用场景

以阿里为例：

Flink角色分工

Flink安装部署方式

1.Local：直接在 IDE 中运行 Flink Job 时则会在本地启动一个 mini Flink 集群。

2.Standalone：在 Flink 目录下执行 bin/start-cluster.sh 脚本则会启动一个 Standalone 模式的集群

3.YARN：YARN 是 Hadoop 集群的资源管理系统，它可以在群集上运行各种分布式应用程序，Flink 可与其他应用并行于 YARN 中，Flink on YARN 的架构如下：

执行流程：

1.Client上传jar包和配置文件到HDFS集群上

2.Client向Yarn ResourceManager提交任务并申请资源

3.ResourceManager分配Container资源并启动ApplicationMaster,然后AppMaster加载Flink的Jar包和配置构建环境,启动JobManager

4.ApplicationMaster向ResourceManager申请工作资源,NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager

5.TaskManager启动后向JobManager发送心跳包，并等待JobManager向其分配任务

On Yarn的两种模式

1.Session会话模式-适合小任务

优点:可以重复使用在Yarn上开启的Flink集群

缺点:适合小任务,不适合大任务

2.Job分离模式-适合大任务

优点:每个任务单独在Yarn上启动一套Flink集群,适合大任务!

缺点:需要频繁的在Yanr上开启Flink集群

Flink架构中常用术语概念

DataStream：

数据流，是指Flink系统处理的最小数据单元。该数据单元最初由外部系统导入，可以通过Socket、Kafka和文件等形式导入，在Flink系统处理后，在通过Socket、Kafka和文件等输出到外部系统，这是Flink的核心概念。

Data Transformation

数据处理单元，会将一或多个DataStream转换成一个新的DataStream。

具体可以细分如下几类：

•一对一的转换：如Map。

•一对0、1或多个的转换：如FlatMap。

•一对0或1的转换，如Filter。

•多对1转换，如Union。

•多个聚合的转换，如window、keyby。

CheckPoint：

CheckPoint是Flink数据处理高可靠、最重要的机制。该机制可以保证应用在运行过程中出现失败时，应用的所有状态能够从某一个检查点恢复，保证数据仅被处理一次（Exactly Once）。

SavePoint：

Savepoint是指允许用户在持久化存储中保存某个checkpoint，以便用户可以暂停自己的任务进行升级。升级完后将任务状态设置为savepoint存储的状态开始恢复运行，保证数据处理的延续性。

FlinkAPI

• Flink API 最底层的抽象为有状态实时流处理。其抽象实现是 Process Function，并且 Process Function 被 Flink 框架集成到了 DataStream API 中来为我们使用。它允许用户在应用程序中自由地处理来自单流或多流的事件（数据），并提供具有全局一致性和容错保障的状态。此外，用户可以在此层抽象中注册事件时间（event time）和处理时间（processing time）回调方法，从而允许程序可以实现复杂计算。

• Flink API 第二层抽象是 Core APIs。实际上，许多应用程序不需要使用到上述最底层抽象的 API，而是可以使用 Core APIs 进行编程：其中包含 DataStream API（应用于有界/无界数据流场景）和 DataSet API（应用于有界数据集场景）两部分。Core APIs 提供的流式 API（Fluent API）为数据处理提供了通用的模块组件，例如各种形式的用户自定义转换（transformations）、联接（joins）、聚合（aggregations）、窗口（windows）和状态（state） *** 作等。此层 API 中处理的数据类型在每种编程语言中都有其对应的类。

• Process Function 这类底层抽象和 DataStream API 的相互集成使得用户可以选择使用更底层的抽象 API 来实现自己的需求。DataSet API 还额外提供了一些原语，比如循环/迭代（loop/iteration） *** 作。

• Flink API 第三层抽象是 Table API。Table API 是以表（Table）为中心的声明式编程（DSL）API，例如在流式数据场景下，它可以表示一张正在动态改变的表。Table API 遵循（扩展）关系模型：即表拥有 schema（类似于关系型数据库中的 schema），并且 Table API 也提供了类似于关系模型中的 *** 作，比如 select、project、join、group-by 和 aggregate 等。Table API 程序是以声明的方式定义应执行的逻辑 *** 作，而不是确切地指定程序应该执行的代码。尽管 Table API 使用起来很简洁并且可以由各种类型的用户自定义函数扩展功能，但还是比 Core API 的表达能力差。此外，Table API 程序在执行之前还会使用优化器中的优化规则对用户编写的表达式进行优化。

• 表和 DataStream/DataSet 可以进行无缝切换，Flink 允许用户在编写应用程序时将 Table API 与 DataStream/DataSet API 混合使用。

• Flink API 最顶层抽象是 SQL。这层抽象在语义和程序表达式上都类似于 Table API，但是其程序实现都是 SQL 查询表达式。SQL 抽象与 Table API 抽象之间的关联是非常紧密的，并且 SQL 查询语句可以在 Table API 中定义的表上执行。

Flink编程模型

总的来说，程序结构分为三部分：数据源>>数据转换>>数据下沉

DataFlow模型

1.Dataflow:Flink程序在执行的时候会被映射成一个数据流模型

2.Operator:数据流模型中的每一个 *** 作被称作Operator,Operator分为:Source/Transform/Sink

3.Partition:数据流模型是分布式的和并行的,执行中会形成1~n个分区

4.Subtask:多个分区任务可以并行,每一个都是独立运行在一个线程中的,也就是一个Subtask子任务

5.Parallelism:并行度,就是可以同时真正执行的子任务数/分区数

Operator传递模式以及Operator Chain

1.One to One模式：一对一模式:分区内的数据是一一对应的,也就是没有shuffle

2.Redistributing 模式：重新分发模式,分区内的数据会重新分布,也就是有shuffle

客户端在提交任务的时候会对Operator进行优化 *** 作，能进行合并的Operator会被合并为一个OperatorChain，实际上就是一个执行链，每个执行链会在TaskManager上一个独立的线程中执行就是SubTask。只有OneToOne模式的才可以合并!

Flink批处理API

Source：

基于集合的:

env.fromElements(可变参数)

env.fromCollection(集合)

env.generateSequence(from,end)

基于文件的

env.readTextFile(“本地/HDFS文件/压缩包/目录”)

env.readCsvFile(“csv格式的文件”)

自定义数据源：kafka、Mysql…

Transformation：

map flatmap filter groupBy sum min max aggregate union distinct join corss reblance PartitionByHash…

Sink:

控制台、文本文件、Kafka、Mysql、Redis…

Flink流处理API

流处理的API与批处理类似，许多都通用，下面列举一些transformation中有区别的：

1.批处理的分组是groupBy,流处理的分组是keyBy

2.处理中不光有union合并还有connect连接,区别如下:

union合并同类型的多个流数据

connect可以连接不同类型的两个流

3.Split就是将一个流分成多个流

4.Select就是获取分流后对应的数据

Flink常用算子介绍

1.Map：

map算子对一个DataStream中的每个元素使用用户自定义的Mapper函数进行处理，每个输入元素对应一个输出元素，最终整个数据流被转换成一个新的DataStream。输出的数据流DataStream类型可能和输入的数据流DataStream不同。

2.Filter：

filter算子对每个元素进行过滤，过滤的过程使用一个Filter函数进行逻辑判断。对于输入的每个元素，如果filter函数返回True，则保留，如果返回False，

3.FlatMap：

flatMap算子和map有些相似，输入都是数据流中的每个元素，与之不同的是，flatMap的输出可以是零个、一个或多个元素，当输出元素是一个列表时，flatMap会将列表展平。

我们可以用切水果的例子来理解map和flatMap的区别。map会对每个输入元素生成一个对应的输出元素：

{苹果，梨，香蕉}.map(去皮) => {去皮苹果, 去皮梨，去皮香蕉}

flatMap先对每个元素进行相应的 *** 作，生成一个相应的集合，再将集合展平：

{苹果，梨，香蕉}.flMap(切碎)

=>

{[苹果碎片1, 苹果碎片2], [梨碎片1，梨碎片2, 梨碎片3]，[香蕉碎片1]}

=>

{苹果碎片1, 苹果碎片2, 梨碎片1，梨碎片2, 梨碎片3，香蕉碎片1}

Flink累加器

Flink累加器：

Flink中的累加器，与Mapreduce counter的应用场景类似，可以很好地观察task在运行期间的数据变化，如在Flink job任务中的算子函数中 *** 作累加器，在任务执行结束之后才能获得累加器的最终结果。

Flink有以下内置累加器，每个累加器都实现了Accumulator接口。

IntCounter、LongCounter、DoubleCounter

编码步骤:

1.创建累加器

private IntCounter numLines = new IntCounter();

2.注册累加器

getRuntimeContext().addAccumulator(“num-lines”, this.numLines);

3.使用累加器

this.numLines.add(1);

4.获取累加器的结果

myJobExecutionResult.getAccumulatorResult(“num-lines”)

Flink广播变量

Flink支持广播。可以将数据广播到TaskManager上就可以供TaskManager中的SubTask/task去使用，数据存储到内存中。这样可以减少大量的shuffle *** 作，而不需要多次传递给集群节点；

比如在数据join阶段，不可避免的就是大量的shuffle *** 作，我们可以把其中一个dataSet广播出去，一直加载到taskManager的内存中，可以直接在内存中拿数据，避免了大量的shuffle，导致集群性能下降；

广播数据：

.withBroadcastSet(DataSet, “name”);

获取广播的数据：

Collection<> broadcastSet = getRuntimeContext().getBroadcastVariable(“name”);

Flink四大基石

FlinkWindow

Window分类：

按照时间和数量划分：

**time-window:时间窗口:**根据时间划分窗口,如:每xx分钟统计最近xx分钟的数据

**count-window:数量窗口:**根据数量划分窗口,如:每xx个数据统计最近xx个数据

按照slide滑动间隔和size大小分类：

窗口有两个重要的属性: 窗口大小size和滑动间隔slide,根据它们的大小关系可分为:

**tumbling-window:滚动窗口:**size=slide,如:每隔10s统计最近10s的数据

**sliding-window:滑动窗口:**size>slide,如:每隔5s统计最近10s的数据

当size

当然还有会话窗口，不太常用，此处省略。

Flink-Time与watermark

EventTime:事件时间:是数据/事件真真正正产生或发生的时间

IngestionTime:摄入时间:是数据/事件到达流处理系统的时间

ProcessingTime:处理时间:是被流系统处理计算的时候的时间

什么是Watermark？

用以衡量事件时间进展。Watermark 也是一种平衡处理延时和完整性的灵活机制

其实Watermark就是给数据再额外的加的一个时间列

Watermark的计算：

Watermark = 数据的事件时间 - 最大允许的延迟时间或乱序时间

FlinkState与Checkpoint

State分类：

有状态计算：相同的输入得到不同的输出/不一定得到相同的输出,就是有状态计算

无状态计算：相同的输入得到相同的输出就是无状态计算

无状态计算的场景:各种简单的转换/过滤等 *** 作,如简单的map/flatMap/filter

有状态计算的场景:如各种聚合统计,sum/reduce/max/min

Checkpoint：

某一时刻,Flink中所有的Operator的当前State的全局快照,一般存在磁盘上

Checkpoint底层使用了Chandy-Lamport 分布式快照算法，可以保证数据的在分布式环境下的一致性