一、Flink介绍

Apache Flink是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。
以内存执行速度和任意规模来执行计算。

中文文档：https://nightlies.apache.org/flink/flink-docs-stable/zh/

Flink有四个基石：Checkpoint机制（保障数据一致性）、State（提供API）、Time（数据容错）、Window（操作窗口）

1.1 Flink简介

1.1.1 Checkpoint
Checkpoint机制 为Flink实现了一个分布式的一致性的快照，从而提供了一致性的语义；

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

1.1.3 Time
Flink还实现了Watemark的机制，能够支持基于事件的时间的处理，或者说基于系统时间的处理，能够容忍数据的延时、容忍数据的迟到、容忍乱序的数据。

1.1.4 Window
Flink提供了开箱即用的各种窗口，比如滑动窗口、滚动窗口、会话窗口以及非常灵活的自定义窗口。

1.2 Flink流处理特性

支持高吞吐、低延迟、高性能的流处理
支持带有事件时间的窗口（Window）操作
支持有状态计算的 Exactly-once 语义
支持高度灵活的窗口（Window）操作，支持基于 time、count、session，以及 data-driven 的窗口操作
支持具有 Backpressure 功能的持续流模型
支持基于轻量级分布式快照（Snapshot）实现的容错
一个运行时同时支持 Batch on Streaming 处理和 Streaming 处理
Flink 在 JVM 内部实现了自己的内存管理
支持迭代计算
支持程序自动优化：避免特定情况下 Shuffle、排序等昂贵操作，中间结果有必要进行缓存

1.3 Flink的批处理和流处理

批处理：有界、持久、大量，批处理适合需要访问全套记录才能完成的计算工作，一般用于离线统计；
流处理：无界、实时，流处理无需针对整个数据集执行操作，而是对通过系统传输的每个数据项进行操作，一般用于实时统计；

Flink可以同时处理批处理和流处理。Flink通过将批处理（即处理有限的静态数据）视作一种特殊的流处理；

1.4 Flink Runtime执行引擎

Flink的核心计算架构是下图中的Flink Runtime执行引擎，他是一个分布式系统，能够接受数据流程序 并在一台或多台机器上以容错方式执行。

上图为Flink技术栈的核心组成部分，值得一提的是，Flink分别提供：
①面向流处理的接口（DataStream API）
②面向批处理的接口（DataSet API）
因此，Flink既可以完成流处理，也可以完成批处理。

Flink还支持的拓展库涉及：
①机器学习（FlinkML）
②复杂事件处理（CEP）
③图计算（Gelly）
④分别针对流处理和批处理的Table API

一一一一一
DataStream API可以流程地分析无限数据流，并且可以用Java或者Scala等来实现。开发者需要基于一个叫DataStream的数据结构来开发，这个数据结构用于表示永不停止的分布式数据流。

Flink的分布式特点体现在他能够在成百上千台机器上运行，它将大型的计算任务分成许多小的部分，每个机器执行一部分。
Flink能够自动地确保发生机器故障或者其他错误时计算能够持续进行，或者在修复bug或进行版本升级后有计划地再次执行一次。这种能力使得开发人员不需要担心运行失败。

Flink本质上使用容错性数据流，这使得开发人员可以分析持续生成切用不结束的数据（即流处理）；

二、Flink运行架构

2.1 Flink程序结构

Flink程序的基本构建块是 ：流和转换；（注意:DataSet API中使用的DataSet也是内部流）

从概念上讲，
流：可能永无止境的数据记录流；
转换：是将一个或多个流作为一个或多个流的操作。（即，输入并产生一个或多个输出流）

2.1.1 Source数据源
Flink在流处理和批处理上的source大概有四类：
① 基于本地集合的source；
② 基于文件的source；
③ 基于网络套接字的source；
④ 自定义的source；

自定义的source常见的有Apache kafka、RabbitMQ等，当然你也可以自定义自己的source；

2.1.2 Transformation
数据转换的各种操作，有Map/FlatMap/Filter/KetBy/Reduce/Fold/Aggregations/Window/WindowAll./Window join/Split/Select等，操作很多，可以将数据转换计算成你想要的数据。

2.1.3Sink
Sink接收器，Flink将转换计算后的数据发送的地点，可能需要存储下来。
Flink常见的Sink大概有如下几类：写入文件、打印出来、写入socket、自定义的sink。

自定义的sink常见的有Apache Kafka、RabbitMQ、MySQL、ElasticSearch、Apache Cassandra、Hadoop FileSystem等，同理也可以定义自己的sink。

2.2 Flink并行数据流

Flink程序本质上是并行的和分布式的；

Flink程序在启动、执行、结束的各个流程：
① 执行的时候，会被映射成一个Streaming Dataflow，一个Streaming Dataflow 是由一组Stream和Transformation Operator组成的。
② 启动时从一个或多个Source Operator开始
③ 结束于一个或多个Sink Operator。

Flink程序执行流程：执行的时候，会被映射成一个Streaming Dataflow，一个Streaming Dataflow 是由一组Stream和Transformation Operator组成的；
一个流Stream 包含一个或多个流分区，而每一个operator包含一个或多个operator子任务，也就是 operator subtask【操作子任务】。
operator subtask【操作子任务】间彼此独立，在不同的线程中执行，甚至是在不同的机器或不同的容器上。
operator subtask【操作子任务】的数量是这一特定operator的并行度。
相同程序中operator有不同级别的并行度。

一个Stream可以被分成多个Stream的分区，也就是Stream Partition【分区】。
一个Operator也可以被分为多个Operator Subtask【子任务】，如下图；

分析图片：

Source被分为Source1 和 Source2，它们是Source的Operator Subtask
【也就是说Source1和Source2都是Source的 Operator Subtask子任务】。
每一个Operator Subtask都是在不同的线程当中独立执行的；
一个Operator的并行度，就等于Operator Subtask的个数；
下图Source的并行度为2，而一个Stream的并行度就等于它生成的Operator的并行度；
数据在两个 operator 之间传递的时候有两种模式：
①One to One 模式：两个operator用此模式传递的时候，会保持数据的分区数和数据的排序；
如下图中的Source1到IMap1，它就保留的Source的分区特性，以及分区元素处理的有序性。
②Redistributing重新分配模式：这种模式会改变数据的分区数；
每一个Operator subtask【如下图的Source1 和 Source2】会根据选择Transformation把数据
发送到不同目标subtasks， 比如keyBy()会通过hashcode重新分区，broadcase()和rebalance()方法会随机重新分区；

2.3 Task和Operator chain

Flink的所有操作都被称之为Operator，客户端在提交任务的时候会对Operator进行优化操作，能进行合并的Operator会被合并为一个Operator，合并后的Operator称为Operator chain，实际上就是一个执行链，每个执行链会在TaskManager上一个独立的线程中执行。

2.4 任务调度与执行

当Flink执行executor会自动根据程序代码生成DAG数据流图；
ActorSystem创建Actor将数据流图发送给JobManager的Actor；
Jobmanager会不断接受TaskManager的心跳消息，从而可以获取到有效的TaskManager；
JobManager通过调度器在TaskManager中调度执行Task（在Flink中，最小的调度单元就是task，对应就是一个线程）；
在程序运行过程中，task和task之间是可以进行数据传输的；
Job Client【就是上图的Flink程序】：

主要职责是提交任务，提交后可以结束进程，也可以等待结果返回；
Job Client 不是Flink程序执行的内部部分，但它是任务执行的起点；
Job Client负责接收用户的程序代码，然后创建数据流，将数据流提交给Job Manager以便进一步执行。执行完成后，Job Client将结果返回给用户；
JobManager：

主要职责是调度工作并协调任务做检查点；
集群中至少要有一个master，master负责调度task，协调checkpoints和容错；
高可用设置的话可以有多个master，但是要保证一个是leader，其他是standby；
JobManager包含Actor System、Scheduler调度器、CheckPoint协调器 三个重要的组件；
JobManager从客户端【上图的Flink程序】接收到任务后，首先生成优化过的执行计划，再调度到TaskManager中执行；
TaskManager

主要职责是从JobManager处接收任务，并部署和启动任务，接收上游的数据并处理；
TaskManager是在JVM中的一个或多个线程中执行任务的工作节点；
TaskManager在创建之初就设置好了Slot【槽】，每个Slot可以执行一个任务；

2.4.1任务槽和槽共享

每个TaskManager是一个JVM的进程，可以在不同的线程中执行一个或多个子任务。为了控制一个worker能接收多少个task。worker通过task slot【任务槽】来进行控制（一个worker至少有一个task slot）

任务槽：

每个task slot表示TaskManager拥有资源的一个固定大小的子集；
flink将进程的内存进行了划分到多个slot中；
上图中有两个taskManager，每个TaskManager有三个slot，每个slot占1/3的内存；
内存被划分到不同的slot之后可以获得如下好处：
①TaskManager最多能同时并发执行的任务是可以控制的，那就是3个，因为不能超过slot的数量；
②slot有独占的内存空间，这样在一个TaskManager中可以运行多个不同的作业，作业之间不受影响；
槽共享：

只需计算Job中最高并行度（parallelism）的task slot，只要这个满足，其他的job也都能满足；
资源分配更加公平，如果有比较空闲的slot可以将更多的任务分配给它。图中若没有任务槽共享，负载不高的Source/Map 等subtask将会占据许多资源，而负载较高的窗口subtask则会缺乏资源；
有了任务槽任务，可以将基本并行度（base parallelism）从2提升到6，提高了分槽资源的利用率。同时还可以保障TaskManager给subtask的分配的slot方案更加公平；

三、Flink快速上手

3.1 准备

需求：统计一段文字中，每个单词出现的频次；
版本：基于1.17.0版本；
数据准备：在工程根目录下创建一个input文件夹，并且在下面创建文本文件words.txt，随便如下数据
批处理思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次；

3.2 导入依赖

org.apache.flink
flink-clients
1.17.0



org.apache.flink
flink-streaming-java
1.17.0

3.3 数据准备

查找到数据源：

思路：流处理，所以是一行一行的读取，然后按照空格切分，然后再分组统计；

3.4、批处理代码 ：实现wordcount 案例

DataSet API 批处理 （过时了）

package com.flink17.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * DataSet API 实现wordcount
 *
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 16:27
 */
public class WordCountStreamDemoMain{
    public static void main(String[] args) throws Exception {
        // 1.创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 2.读取数据（这里是文本数据）
        DataSource lineDS = env.readTextFile("input/words.txt");
        // 3.切分、转换（word,1）的格式（第一个参数 word表示的是单词，第二个参数 1是指出现的次数）
        // flatMap的方法中的FlatMapFunction 的第一个参数是输入，第二个参数是输出
        FlatMapOperator> wordAndOne
                = lineDS.flatMap(new FlatMapFunction>() {
            // 重写 flatMap ，第一个参数是需要操作的数据，第二个参数Collector是采集器
            @Override
            public void flatMap(String s, Collector> collector) throws Exception {
                //3.1 按照空格切割
                String[] words = s.split(" ");
                //3.2 将单词转换成二元组（word，1）这样的格式
                for (String word : words) {
                    Tuple2 wordTuple2 = Tuple2.of(word, 1);
                    // 3.3 使用Collector向下游发送数据
                    collector.collect(wordTuple2);
                }
            }
        });
        // 4.按照word分组
        UnsortedGrouping> wordAndOneGroupBy = wordAndOne.groupBy(0);
        // 5.各分组内聚合
        AggregateOperator> sum =
                wordAndOneGroupBy.sum(1); // 这里的1 是位置，表示第二个元素
        // 6.输出
        sum.print();
    }
}

执行结果

DataSet写法实现批处理是过时的，推荐使用DataStream来写；

3.5、流处理代码（有界流-有开始有结束）：：实现wordcount 案例

package com.flink17.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * DataStream流处理 实现wordcount：（读文件，有界流）
 *
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 16:53
 */
public class WordCountStreamDemoMain {
    public static void main(String[] args) throws Exception {
        // 1.创建执行环境
        StreamExecutionEnvironment evn = StreamExecutionEnvironment.getExecutionEnvironment();
        // 2.读取数据
        DataStreamSource lineDs = evn.readTextFile("input/words.txt");
        // 3.处理数据:切分、转换、分组、聚合
        SingleOutputStreamOperator> wordAndOne = lineDs.flatMap(new FlatMapFunction>() {
            @Override
            public void flatMap(String value, Collector> out) throws Exception {
                String[] words = value.split(" "); // 空格切分
                for (String word : words) {
                    // 转换成二元组
                    Tuple2 wordsAndOne = Tuple2.of(word, 1);
                    // 使用Collector向下游发送数据
                    out.collect(wordsAndOne);
                }
            }
        });
        // 分组
        KeyedStream, String> ks = wordAndOne.keyBy(
                // 第一个参数书数据的类型，第二个参数是key的类型
                new KeySelector, String>() {
                    @Override
                    public String getKey(Tuple2 value) throws Exception {
                        return value.f0;
                    }
                }
        );
        SingleOutputStreamOperator> sum = ks.sum(1);// 这里的1 是位置，表示第二个元素
        // 4.输出数据
        sum.print();
        // 5.执行
        evn.execute();
    }
}

执行结果

流处理：拿一条处理一条；所以是一行一行的读取；且有状态，比如hello最后为（hello，3）3是有状态的计算；
前面的编号就是并行度编号，也就是线程数编号；

批处理：一口气处理一批（这里就是整个文本）

3.6、流处理代码

实现wordcount 案例
在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续地处理捕获的数据；
模拟场景：监听socket端口，然后向该端口不断的发送数据

将StreamWorldCount代码中读取文件数据的readTextFile方法替换成socket文本流的方法socketTextStram；

package com.flink17.demo;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author lc
 * @version 1.0
 * @date 2024/10/8 0008 17:17
 */
public class WordCountStreamUnboundDemoMain {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.读取数据：socket,第一个参数书服务器名称，第二个参数是端口号
        DataStreamSource socketDS = env.socketTextStream("hadoop102", 7777);
        //3.处理数据：切分，转换，分组，聚合，输出
        socketDS
                .flatMap(
                        (String value, Collector> out) -> {
                            String[] words = value.split(" "); // 空格切分
                            for (String word : words) {
                                // 转换成二元组
                                Tuple2 wordsAndOne = Tuple2.of(word, 1);
                                // 使用Collector向下游发送数据
                                out.collect(wordsAndOne);
                            }
                        }
                )
                .returns(Types.TUPLE(Types.STRING,Types.INT)) //需要执行返回类型，（word，1）
                .keyBy(value -> value.f0)
                .sum(1)
                .print(); // 输出

        //4.执行
        env.execute();
    }
}

不指定返回类型的化，执行会报错；因为Flink具有一个类型提取系统，可以分析函数的输入和返回类型，自动获取类型信息，从而获得对应序列化器和反序列化器，但是，由于java中泛型查出的存在所以报错了；

执行，就会一直监听那个服务器，有输入才会有响应的输出；

有界流是有结束的，无界留是没有结束的

————————————————