DataStream API （基础篇）

---

注： 本文只涉及DataStream

• 原因：随着大数据和流式计算需求的增长，处理实时数据流变得越来越重要。因此，DataStream由于其处理实时数据流的特性和能力，逐渐替代了DataSet成为了主流的数据处理方式。

目录

DataStream API （基础篇）

前摘：

一、执行环境

1. 创建执行环境

2. 执行模式

3. 触发程序执行

二、源算子（source）

三、转换算子（Transformation）

四、输出算子（sink）

---

前摘：

一个 Flink 程序，其实就是对 DataStream 的各种转换。具体来说，代码基本上都由以下几 部分构成，如图所示：

• 获取执行环境（Execution Environment）
• 读取数据源（Source）
• 定义基于数据的转换操作（Transformations）
• 定义计算结果的输出位置（Sink）
• 触发程序执行（Execute）

其中，获取环境和触发执行，都可以认为是针对执行环境的操作。所以本章我们就从执行 环境、数据源（source）、转换操作（Transformation）、输出（Sink）四大部分，对常用的 DataStream API 做基本介绍。

一、执行环境

1. 创建执行环境

• 编写Flink程序的第一步就是创建执行环境。
• 我 们 要 获 取 的 执 行 环 境 ， 是 StreamExecutionEnvironment 类的对象，这是所有 Flink 程序的基础
• 在代码中创建执行环境的 方式，就是调用这个类的静态方法，具体有以下三种。

1. getExecutionEnvironment
   最简单的方式，就是直接调用 getExecutionEnvironment 方法。它会根据当前运行的上下文 直接得到正确的结果；

   //此处的 env 是 StreamExecutionEnvironment 对象
   val env = StreamExecutionEnvironment.getExecutionEnvironment

2. createLocalEnvironment
   这个方法返回一个本地执行环境。可以在调用时传入一个参数，指定默认的并行度；如果 不传入，则默认并行度就是本地的 CPU 核心数。

   //此处的 localEnvironment 是 StreamExecutionEnvironment 对象
   val localEnvironment = StreamExecutionEnvironment.createLocalEnvironment()
   

3. createRemoteEnvironment
   这个方法返回集群执行环境。需要在调用时指定 JobManager 的主机名和端口号，并指定 要在集群中运行的 Jar 包。

   //此处的 remoteEnv 是 StreamExecutionEnvironment 对象
   val remoteEnv = StreamExecutionEnvironment
    .createRemoteEnvironment(
    "host", // JobManager 主机名
    1234, // JobManager 进程端口号
    "path/to/jarFile.jar" // 提交给 JobManager 的 JAR 包
   )
   

2. 执行模式

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

• 流执行模式（STREAMING） 这是 DataStream API 最经典的模式，一般用于需要持续实时处理的无界数据流。默认情 况下，程序使用的就是 STREAMING 执行模式。
• 批执行模式（BATCH） 专门用于批处理的执行模式, 这种模式下，Flink 处理作业的方式类似于 MapReduce 框架。 对于不会持续计算的有界数据，我们用这种模式处理会更方便。
• 自动模式（AUTOMATIC） 在这种模式下，将由程序根据输入数据源是否有界，来自动选择执行模式

由于 Flink 程序默认是 STREAMING 模式，我们这里重点介绍一下 BATCH 模式的配置。 主要有两种方式：

（1）通过命令行配置

bin/flink run -Dexecution.runtime-mode=BATCH ...

在提交作业时，增加 execution.runtime-mode 参数，指定值为 BATCH。

（2）通过代码配置

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setRuntimeMode(RuntimeExecutionMode.BATCH)

3. 触发程序执行

我们需要显式地调用执行环境的 execute()方法，来触发程序执行。execute()方法将一直等 待作业完成，然后返回一个执行结果（JobExecutionResult）。

env.execute()

二、源算子（source）

Source源算子（基础篇二）

三、转换算子（Transformation）

持续更新中

四、输出算子（sink）

持续更新中