Flink 基础

1.概述

Apache Flink 是一个分布式处理引擎和框架，用于对无界和有界数据流进行有状态的计算。

Flink 主要特点：

• 高吞吐和低延时。每秒处理数百万个事件，毫秒级延迟。
• 结果的准确性。Flink提供了事件时间（event-time）和处理时间（processing-time）语义。对于乱序事件流，事件时间语义仍然能提供一致且准确的结果。
• 精确一次（exactly-once）的状态一致性保证。
• 可以连接到最常用的外部系统，如 Kafka、Hive、JDBC、HDFS、Redis等。
• 高可用。本身高可用的设置，加上与 K8s 、YARN 和 Mesos 的紧密集成，再加上从故障中快速恢复和动态拓展任务的能力，Flink能做到以极少的停机时间7X24全天候运行。

Flink API：

2. 快速上手代码

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class Flink05UnboundStream {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(new TextLineInputFormat(), new Path("input/")).build();
        DataStreamSource<String> dataStreamSource = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "file");
        SingleOutputStreamOperator<Tuple2<String, Integer>> word_one = dataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = s.split(" +");
                for (String word : words) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        });
        word_one.keyBy((KeySelector<Tuple2<String, Integer>, String>) stringIntegerTuple2 -> stringIntegerTuple2.f0).sum(1).print();

        env.execute();
    }
}

3. 作业部署模式（基于 YARN）

3.1 会话模式

会话模式需要启动一个集群，保持一个会话。客户端通过这个会话提交作业。因为集群启动时所有资源已经确定，所以提交的所有作业会竞争集群中的资源。

bin/yarn-session.sh -nm test

可用参数解读：

• -d：分离模式，如果你不想让Flink YARN客户端一直前台运行，可以使用这个参数，即使关掉当前对话窗口，YARN session也可以后台运行。
• -jm（--jobManagerMemory）：配置JobManager所需内存，默认单位MB。
• -nm（--name）：配置在YARN UI界面上显示的任务名。
• -qu（--queue）：指定YARN队列名。
• -tm（--taskManager）：配置每个TaskManager所使用内存。

注意：Flink1.11.0版本不再使用 -n 参数和 -s 参数分别指定 TaskManager 数量和 slot 数量，YARN 会按照需求动态分配 TaskManager 和 slot 。所以从这个意义上讲，YARN的会话模式也不会把集群资源固定，同样是动态分配的。

bin/flink run -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar

3.2 单作业模式

单作业模式为每个提交的作业启动一个集群。Flink 本身无法直接这样运行，需要借助资源管理框架（YARN、Kubernetes）来启动集群。

bin/flink run -d -t yarn-per-job -c com.atguigu.wc.SocketStreamWordCount FlinkTutorial-1.0-SNAPSHOT.jar

bin/flink list -t yarn-per-job -Dyarn.application.id=application_XXXX_YY

bin/flink cancel -t yarn-per-job -Dyarn.application.id=application_XXXX_YY <jobId>

3.3 应用模式

会话模式和单作业模式的应用代码都是在客户端上执行，然后由客户端提交给 JobManager 的。这种方式客户端需要占用大量网络带宽，去下载依赖和把二进制数据发送给 JobManager；加上很多情况下提交作业用的是同一个客户端，就会加重客户端所在节点的资源消耗。所以解决办法就是，直接把应用提交到 JobManger 上运行。所以需要为每一个提交的应用单独启动一个JobManager，也就是创建一个集群。这个JobManager只为执行这一个应用而存在，执行结束之后JobManager也就关闭了，这就是所谓的应用模式。

bin/flink run-application -t yarn-application -Dyarn.provided.lib.dirs="hdfs://hadoop102:8020/flink-dist" -c com.atguigu.wc.SocketStreamWordCount  hdfs://hadoop102:8020/flink-jars/FlinkTutorial-1.0-SNAPSHOT.jar

bin/flink list -t yarn-application -Dyarn.application.id=application_XXXX_YY

bin/flink cancel -t yarn-application -Dyarn.application.id=application_XXXX_YY <jobId>

3.4 三种模式对比

4. Flink 运行时架构

4.1 会话模式系统架构流程

1）作业管理器（JobManager）

JobManager 是一个Flink集群中任务管理和调度的核心，是控制应用执行的主进程。也就是说，每个应用都应该被唯一的 JobManager 所控制执行。

JobManger 又包含3个不同的组件。

（1）JobMaster

JobMaster 是 JobManager 中最核心的组件，负责处理单独的作业（Job）。所以 JobMaster 和具体的 Job 是一一对应的，多个 Job 可以同时运行在一个Flink集群中, 每个 Job 都有一个自己的 JobMaster。

在作业提交时，JobMaster 会先接收到要执行的应用。JobMaster 会把 JobGraph 转换成一个物理层面的数据流图，这个图被叫作“执行图”（ExecutionGraph），它包含了所有可以并发执行的任务。JobMaster会向资源管理器（ResourceManager）发出请求，申请执行任务必要的资源。一旦它获取到了足够的资源，就会将执行图分发到真正运行它们的 TaskManager 上。

（2）资源管理器（ResourceManager）

ResourceManager 主要负责资源的分配和管理，在 Flink 集群中只有一个。所谓“资源”，主要是指 TaskManager 的任务槽（task slots）。任务槽就是Flink集群中的资源调配单元，包含了机器用来执行计算的一组 CPU 和内存资源。每一个任务（Task）都需要分配到一个slot上执行。

（3）分发器（Dispatcher）

Dispatcher 主要负责提供一个 REST 接口，用来提交应用，并且负责为每一个新提交的作业启动一个新的 JobMaster 组件。Dispatcher 也会启动一个 Web UI，用来方便地展示和监控作业执行的信息。

2）任务管理器（TaskManager）

TaskManager 是 Flink 中的工作进程，数据流的具体计算就是它来做的。Flink集群中必须至少有一个 TaskManager；每一个 TaskManager 都包含了一定数量的任务槽（task slots）。Slot 是资源调度的最小单位，slot 的数量限制了TaskManager能够并行处理的任务数量。

启动之后，TaskManager 会向资源管理器注册它的 slots；收到资源管理器的指令后，TaskManager 就会将一个或者多个槽位提供给 JobMaster 调用，JobMaster 就可以分配任务来执行了。在执行过程中，TaskManager 可以缓冲数据，还可以跟其他运行同一应用的 TaskManager 交换数据。

4.2 核心概念

并行度：一个特定算子的子任务（subtask）的个数被称之算子的并行度（parallelism）。

• 代码中设置：算子后跟着调用setParallelism()；执行环境的setParallelism()。注意：keyBy不是算子，所以无法对keyBy设置并行度
• 提交应用时设置：-p 参数来指定当前应用程序执行的并行度
• 配置文件中设置：配置文件flink-conf.yaml -> parallelism.default: 2

算子链：并行度相同的一对一（one to one）算子操作，可以直接链接在一起形成一个“大”的任务（task），每个task会被一个线程执行。Flink默认会按照算子链的原则进行链接合并。

任务槽：Flink 中每一个 TaskManager 都是一个JVM进程，TaskManager的计算资源是有限的，并行的任务越多，每个线程的资源就会越少。TaskManager上对每个任务运行所占用的资源做出明确的划分，这就是所谓的任务槽（task slots）。每个任务槽（task slot）其实表示了 TaskManager 拥有计算资源的一个固定大小的子集。flink-conf.yaml配置文件中 taskmanager.numberOfTaskSlots。需要注意的是，slot目前仅仅用来隔离内存，不会涉及CPU的隔离。

flink run-application -t yarn-application -d -Dyarn.application.name=my-job -Dyarn.application.queue=realtime -Djobmanager.memory.process.size=2048m -Dtaskmanager.memory.process.size=4096m -Dtaskmanager.numberOfTaskSlots=4 -Dparallelism.default=12 -c com.xxx.MyJob /path/to/job.jar

Flink默认是允许slot共享的，如果希望某个算子对应的任务完全独占一个slot，或者只有某一部分算子共享slot，可以通过算子的 slotSharingGroup("组名") slot 共享组手动指定。

任务槽和并行度的关系：任务槽是静态的概念，TaskManager 具有的并发执行能力，可以通过参数 taskmanager.numberOfTaskSlots 进行配置；而并行度是动态概念，也就是 TaskManager 运行程序时实际使用的并发能力，可以通过参数parallelism.default进行配置。举例说明：standalone的会话模式，假设一共有3个TaskManager，每一个TaskManager中的slot数量设置为3个，那么一共有9个task slot，表示集群最多能并行执行9个同一算子的子任务。

4.3 YARN模式系统架构流程

注意：AM 是 YARN 视角的名字，JobManager 是 Flink 视角的名字

5. 代码生成任务的过程