Apache Flink——DataStream API 执行环境
前言
Flink 有非常灵活的分层 API 设计，其中的核心层就是 DataStream/DataSet API。由于新版本已经实现了流批一体，DataSet API 将被弃用，官方推荐统一使用 DataStream API 处理流数据和批数据。
DataStream（数据流）本身是 Flink 中一个用来表示数据集合的类（Class），我们编写的Flink 代码其实就是基于这种数据类型的处理，所以这套核心 API 就以 DataStream 命名。对于批处理和流处理，我们都可以用这同一套 API 来实现。
一个 Flink 程序，其实就是对 DataStream 的各种转换。具体来说，代码基本上都由以下几部分构成：
- 获取执行环境（execution environment）
- 读取数据源（source）
- 定义基于数据的转换操作（transformations）
- 定义计算结果的输出位置（sink）
- 触发程序执行（execute）
其中，获取环境和触发执行，都可以认为是针对执行环境的操作。
一、执行环境（Execution Environment）
Flink 程序可以在各种上下文环境中运行：我们可以在本地 JVM 中执行程序，也可以提交到远程集群上运行。
不同的环境，代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时，首先必须获取当前 Flink 的运行环境，从而建立起与 Flink 框架之间的联系。只有获取了环境上下文信息，才能将具体的任务调度到不同的 TaskManager 执行。
1.1 创建执行环境
- StreamExecutionEnvironment.getExecutionEnvironment()
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
这种“智能”的方式不需要我们额外做判断，用起来简单高效，是最常用的一种创建执行环境的方式。
1.2 StreamExecutionEnvironment.createLocalEnvironment()
这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果不传入，则默认并行度就是本地的 CPU 核心数。
StreamExecutionEnvironment localEnv = StreamExecutionEnvironment.createLocalEnvironment();
1.3 StreamExecutionEnvironment.createRemoteEnvironment()
这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定要在集群中运行的 Jar 包。
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
"host", // JobManager 主机名
1234, // JobManager 进程端口号
"path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
);
二、执行模式(Execution Mode)
我们获取到的执行环境，是一个 StreamExecutionEnvironment，顾名思义它应该是做流处理的。那对于批处理，又应该怎么获取执行环境呢？
在之前的 Flink 版本中，批处理的执行环境与流处理类似，是调用类ExecutionEnvironment的静态方法，返回它的对象：
// 批处理环境
ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();
// 流处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
而从 1.12.0 版本起，Flink 实现了 API 上的流批统一。DataStream API 新增了一个重要特性：可以支持不同的“执行模式”（execution mode），通过简单的设置就可以让一段 Flink 程序在流处理和批处理之间切换。这样一来，DataSet API 也就没有存在的必要了。
-
流执行模式（STREAMING）
这是 DataStream API 最经典的模式，一般用于需要持续实时处理的无界数据流。默认情况下，程序使用的就是 STREAMING 执行模式。 -
批执行模式（BATCH）
专门用于批处理的执行模式, 这种模式下，Flink 处理作业的方式类似于MapReduce 框架。对于不会持续计算的有界数据，我们用这种模式处理会更方便。 -
自动模式（AUTOMATIC）
在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式。
2.1 BATCH 模式的配置方法
由于 Flink 程序默认是 STREAMING 模式，我们这里重点介绍一下 BATCH 模式的配置。
主要有两种方式：
- 1、通过命令行配置
bin/flink run -Dexecution.runtime-mode=BATCH
在提交作业时，增加 execution.runtime-mode 参数，指定值为 BATCH。
- 2、通过代码配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
在代码中，直接基于执行环境调用 setRuntimeMode 方法，传入 BATCH 模式。
三、触发程序执行
有了执行环境，我们就可以构建程序的处理流程了：基于环境读取数据源，进而进行各种转换操作，最后输出结果到外部系统。
需要注意的是，写完输出（sink）操作并不代表程序已经结束。因为当 main()方法被调用时，其实只是定义了作业的每个执行操作，然后添加到数据流图中；这时并没有真正处理数据——因为数据可能还没来。Flink 是由事件驱动的，只有等到数据到来，才会触发真正的计算，这也被称为“延迟执行”或“懒执行”（lazy execution）。
所以我们需要显式地调用执行环境的 execute()方法，来触发程序执行。execute()方法将一直等待作业完成，然后返回一个执行结果（JobExecutionResult）。
env.execute();
参考：
https://blog.csdn.net/weixin_45417821/article/details/124141186
共有 0 条评论