Flink API

imreW / 2023-07-04 / 原文

Flink 的 DataSet 和 DataStream 的 API，并模拟了实时计算的场景。

说好的流批一体呢

现状

Flink 很重要的一个特点是“流批一体”，然而事实上 Flink 并没有完全做到所谓的“流批一体”，即编写一套代码，可以同时支持流式计算场景和批量计算的场景。目前截止 1.10 版本依然采用了 DataSet 和 DataStream 两套 API 来适配不同的应用场景。

DateSet 和 DataStream 的区别和联系

在官网或者其他网站上，都可以找到目前 Flink 支持两套 API 和一些应用场景，但大都缺少了“为什么”这样的思考。

Apache Flink 在诞生之初的设计哲学是：用同一个引擎支持多种形式的计算，包括批处理、流处理和机器学习等。尤其是在流式计算方面，Flink 实现了计算引擎级别的流批一体。那么对于普通开发者而言，如果使用原生的 Flink ，直接的感受还是要编写两套代码。

整体架构如下图所示：

在 Flink 的源代码中，我们可以在 flink-java 这个模块中找到所有关于 DataSet 的核心类，DataStream 的核心实现类则在 flink-streaming-java 这个模块。

在上述两张图中，我们分别打开 DataSet 和 DataStream 这两个类，可以发现，二者支持的 API 都非常丰富且十分类似，比如常用的 map、filter、join 等常见的 transformation 函数。

对于 DataSet 而言，Source 部分来源于文件、表或者 Java 集合；而 DataStream 的 Source 部分则一般是消息中间件比如 Kafka 等。

（一）Environment

Flink有以下几种Environment

1. 批处理Environment，ExecutionEnvironment

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

2.流处理Environment，StreamExecutionEnvironment

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

3. 本机Environment，LocalEnvironment

ExecutionEnvironment env = LocalEnvironment.getExecutionEnvironment();

4. java集合Environment，CollectionEnvironment

ExecutionEnvironment env = CollectionEnvironment.getExecutionEnvironment();

创建Environment的方法

1. getExecutionEnvironment ，含义就是本地运行就是 createLocalEnvironment，如果是通过client提交到集群上，就返回集群的环境

Creates an execution environment that represents the context in which the program is currently executed.
    * If the program is invoked standalone, this method returns a local execution environment, as returned by
    * {@link #createLocalEnvironment()}. If the program is invoked from within the command line client to be
    * submitted to a cluster, this method returns the execution environment of this cluster.

2. createLocalEnvironment ，返回本地执行环境，需要在调用时指定默认的并行度，比如

LocalStreamEnvironment env1 = StreamExecutionEnvironment.createLocalEnvironment(1);
 
LocalEnvironment env2 = ExecutionEnvironment.createLocalEnvironment(1);

3. createRemoteEnvironment，返回集群执行环境，将 Jar 提交到远程服务器。需要在调用时指定 JobManager 的 IP 和端口号，并指定要在集群中运行的 Jar 包，比如

StreamExecutionEnvironment env1 = StreamExecutionEnvironment.createRemoteEnvironment("127.0.0.1", 8080, "/path/word_count.jar");
 
ExecutionEnvironment env2 = ExecutionEnvironment.createRemoteEnvironment("127.0.0.1", 8080, "/path/word_count.jar");

在工作中我们一般使用IntelliJ IDEA开发工具进行代码开发，为了能方便快速的调试Flink和了解Flink程序的运行情况，我们希望本地开发工具中运行Flink时能查看到WebUI，这就可以在编写Flink程序时开启本地WebUI。

在Flink1.15版本之前根据使用Scala版本在Java Flink项目或Scala Flink项目中添加对应Scala版本的依赖。在Flink1.15版本之后，无论是Java Flink项目还是Scala Flink项目，添加如下依赖，不需额外依赖Scala版本。

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-runtime-web</artifactId>
  <version>${flink.version}</version>
</dependency>

Flink Java 代码启动本地WebUI：

Configuration conf = new Configuration();
//设置WebUI绑定的本地端口
conf.setString(RestOptions.BIND_PORT,"8081");
//使用配置
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

执行

env.execute();

（二）DataStream API

DataStream是Flink编写流处理作业的API。我们前面说过一个完整的Flink处理程序应该包含三部分：数据源(Source)、转换操作(Transformation)、结果接收(Sink)。下面我们从这三部分来看DataStream API。

（三）数据源(Source)

（四）数据转换(Transformation)

disableChaining：针对某一个算子操作，断开算子前后的Operator chain，使算子单独作为一个subtask。

startNewChain：从当前算子之前开始划分一条新的Operator chain，在算子之后调用；

 SingleOutputStreamOperator<Tuple2<String,Integer>> sum = socketDS
                .flatMap(
                        (String value, Collector<String> out) -> {
                            String[] words = value.split(" ");
                            for (String word : words) {
                                out.collect(word);
                            }
                        }
                )
                .startNewChain()
                .returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(value -> value.f0)
                .sum(1);