目录

1.写在前面

2.Environment&＃xff1a;执行环境

2.1 getExecutionEnvironment&＃xff1a;得到执行环境&＃xff08;常用方法&＃xff09;

2.2 createLocalEnvironment&＃xff1a;创建本地执行环境

2.3 createRemoteEnvironment&＃xff1a;创建远程执行环境

3.执行模式&＃xff08;Flink 1.12.0&＃xff09;

3.1 流执行模式&＃xff08;STREAMING&＃xff09;

3.2 批执行模式&＃xff08;BATCH&＃xff09;

3.3 自动模式&＃xff08;AUTOMATIC&＃xff09;

4.触发程序执行

1.写在前面

        Flink 有非常灵活的分层 API 设计&＃xff0c;其中的核心层就是 DataStream/DataSet API。由于新版 本已经实现了流批一体&＃xff0c;DataSet API 将被弃用&＃xff0c;官方推荐统一使用 DataStream API 处理流数 据和批数据。我们主要研究dataStream的api&＃xff0c;dataSet与之类似。

        DataStream&＃xff08;数据流&＃xff09;本身是 Flink 中一个用来表示数据集合的类&＃xff08;Class&＃xff09;&＃xff0c;我们编写的 Flink 代码其实就是基于这种数据类型的处理&＃xff0c;所以这套核心 API 就以 DataStream 命名。对于 批处理和流处理&＃xff0c;我们都可以用这同一套 API 来实现。 DataStream 在用法上有些类似于常规的 Java 集合&＃xff0c;但又有所不同。我们在代码中往往并 不关心集合中具体的数据&＃xff0c;而只是用 API 定义出一连串的操作来处理它们&＃xff1b;这就叫作数据流 的“转换”&＃xff08;transformations&＃xff09;。

        一个 Flink 程序&＃xff0c;其实就是对 DataStream 的各种转换。具体来说&＃xff0c;代码基本上都由以下几 69 部分构成。source用来读取数据源&＃xff0c;transform做转换计算&＃xff0c;sink阶段主要用来输出计算数据。在这三个步骤之前&＃xff0c;我们需要创建flink的执行环境。

• 获取执行环境&＃xff08;execution environment&＃xff09;
• 读取数据源&＃xff08;source&＃xff09;
• 定义基于数据的转换操作&＃xff08;transformations&＃xff09;
• 定义计算结果的输出位置&＃xff08;sink&＃xff09;
• 触发程序执行&＃xff08;execute&＃xff09; 

2.Environment&＃xff1a;执行环境

        Flink 程序可以在各种上下文环境中运行&＃xff1a;我们可以在本地 JVM 中执行程序&＃xff0c;也可以提交 到远程集群上运行。 不同的环境&＃xff0c;代码的提交运行的过程会有所不同。这就要求我们在提交作业执行计算时&＃xff0c; 首先必须获取当前 Flink 的运行环境&＃xff0c;从而建立起与 Flink 框架之间的联系。只有获取了环境 上下文信息&＃xff0c;才能将具体的任务调度到不同的 TaskManager 执行。

2.1 getExecutionEnvironment&＃xff1a;得到执行环境&＃xff08;常用方法&＃xff09;

        创建一个执行环境&＃xff0c;表示当前执行程序的上下文。 如果程序是独立调用的&＃xff0c;则 此方法返回本地执行环境&＃xff1b;如果从命令行客户端调用程序以提交到集群&＃xff0c;则此方法返回此集群的执行环境&＃xff0c;也就是说&＃xff0c;getExecutionEnvironment 会根据查询运行的方式决定返回什么样的运行环境&＃xff0c;是最常用的一种创建执行环境的方式。这个方法会根据当前运行的方式&＃xff0c;自行决定该返回什么样的运行环境。

//批处理
ExecutionEnvironment env &＃61; ExecutionEnvironment.getExecutionEnvironment();

//流处理
StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();

        如果没有设置并行度&＃xff0c;会以 flink-conf.yaml 中的配置为准&＃xff0c;默认是 1&＃xff0c;当然我们也可以设置并行度。 

2.2 createLocalEnvironment&＃xff1a;创建本地执行环境

        返回本地执行环境&＃xff0c;需要在调用时指定默认的并行度。如果不传入&＃xff0c;则默认并行度就是本地的 CPU 核心数。

//流处理
LocalStreamEnvironment env &＃61; StreamExecutionEnvironment.createLocalEnvironment(1);

2.3 createRemoteEnvironment&＃xff1a;创建远程执行环境

        返回集群执行环境&＃xff0c;将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号&＃xff0c;并指定要在集群中运行的 Jar 包。

StreamExecutionEnvironment remoteEnv &＃61; StreamExecutionEnvironment.createRemoteEnvironment("host", // JobManager 主机名1234, // JobManager 进程端口号"path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
);

StreamExecutionEnvironment env &＃61;
StreamExecutionEnvironment.createRemoteEnvironment("jobmanage-hostname", 6123, "YOURPATH//WordCount.jar");

        在获取到程序执行环境后&＃xff0c;我们还可以对执行环境进行灵活的设置。比如可以全局设置程 序的并行度、禁用算子链&＃xff0c;还可以定义程序的时间语义、配置容错机制。

3.执行模式&＃xff08;Flink 1.12.0&＃xff09;

        从 1.12.0 版本起&＃xff0c;Flink 实现了 API 上的流批统一。DataStream API 新增了一个重要特 性&＃xff1a;可以支持不同的“执行模式”&＃xff08;execution mode&＃xff09;&＃xff0c;通过简单的设置就可以让一段 Flink 程序 在流处理和批处理之间切换。这样一来&＃xff0c;DataSet API 也就没有存在的必要了。

3.1 流执行模式&＃xff08;STREAMING&＃xff09;

        这是 DataStream API 最经典的模式&＃xff0c;一般用于需要持续实时处理的无界数据流。默认情 71 况下&＃xff0c;程序使用的就是 STREAMING 执行模式。

3.2 批执行模式&＃xff08;BATCH&＃xff09;

        专门用于批处理的执行模式, 这种模式下&＃xff0c;Flink 处理作业的方式类似于 MapReduce 框架。 对于不会持续计算的有界数据&＃xff0c;我们用这种模式处理会更方便。

        Flink 程序默认是 STREAMING 模式&＃xff0c;那么怎么使用batch模式呢&＃xff1f;有以下两种方式&＃xff1a;

&＃xff08;1&＃xff09;通过命令行配置&＃xff0c;在提交作业时&＃xff0c;增加 execution.runtime-mode 参数&＃xff0c;指定值为 BATCH。

bin/flink run -Dexecution.runtime-mode&＃61;BATCH ...

&＃xff08;2&＃xff09; 通过代码配置&＃xff0c;推荐&＃xff0c;因为扩展性更好&＃xff0c;更加灵活

StreamExecutionEnvironment env &＃61; StreamExecutionEnvironment.getExecutionEnvironment();
env.setRuntimeMode(RuntimeExecutionMode.BATCH);


3.3 自动模式&＃xff08;AUTOMATIC&＃xff09;

在这种模式下&＃xff0c;将由程序根据输入数据源是否有界&＃xff0c;来自动选择执行模式。、

4.触发程序执行

        有了执行环境&＃xff0c;我们就可以构建程序的处理流程了&＃xff1a;基于环境读取数据源&＃xff0c;进而进行各种 转换操作&＃xff0c;最后输出结果到外部系统。 需要注意的是&＃xff0c;写完输出&＃xff08;sink&＃xff09;操作并不代表程序已经结束。因为当 main()方法被调用时&＃xff0c;其实只是定义了作业的每个执行操作&＃xff0c;然后添加到数据流图中&＃xff1b;这时并没有真正处理数据 ——因为数据可能还没来。

        Flink 是由事件驱动的&＃xff0c;只有等到数据到来&＃xff0c;才会触发真正的计算&＃xff0c; 这也被称为“延迟执行”或“懒执行”&＃xff08;lazy execution&＃xff09;。 所以我们需要显式地调用执行环境的 execute()方法&＃xff0c;来触发程序执行。execute()方法将一 直等待作业完成&＃xff0c;然后返回一个执行结果&＃xff08;JobExecutionResult&＃xff09;。

env.execute();