Flink DataStream API-526互联

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)

Flink应用程序从数据源获取要处理的数据，DataStream通过StreamExecutionEnvironment.addResource(SourceFunction) 来添加数据源。为了方便使用，Flink预提几类预定义的数据源，比如读取文件的Source、通过Sockt读取的Source、从内存中获取的Source等。

1.基于集合的预定义Source

基于集合的数据源一般是指从内存集合中直接读取要处理的数据，StreamExecutionEnvironment提供了4类预定义方法。

fromCollection

fromCollection是从给定的集合中创建DataStream，StreamExecutionEnvironment提供了4种重载方法：

fromCollection(Collection<T> data)：通过给定的集合创建DataStream。返回数据类型为集合元素类型。
fromCollection(Collection<T> data,TypeInformation<T> typeInfo)：通过给定的非空集合创建DataStream。返回数据类型为typeInfo。
fromCollection(Iterator<T> data,Class<T> type)：通过给定的迭代器创建DataStream。返回数据类型为type。
fromCollection(Iterator<T> data,TypeInformation<T> typeInfo)：通过给定的迭代器创建DataStream。返回数据类型为typeInfo。

fromParallelCollection

fromParallelCollection和fromCollection类似，但是是并行的从迭代器中创建DataStream。

fromParallelCollection(SplittableIterator<T> data,Class<T> type)
fromParallelCollection(SplittableIterator<T>,TypeInfomation typeInfo)

和Iterable中Spliterator类似，这是JDK1.8新增的特性，并行读取集合元素。

fromElements

fromElements从给定的对象序列中创建DataStream，StreamExecutionEnvironment提供了2种重载方法：

fromElements(T... data)：从给定对象序列中创建DataStream，返回的数据类型为该对象类型自身。
fromElements(Class<T> type,T... data)：从给定对象序列中创建DataStream，返回的数据类型type。

generateSequence

generateSequence(long from,long to)从给定间隔的数字序列中创建DataStream，比如from为1，to为10，则会生成1~10的序列。

2.基于Socket的预定义Source

我们还可以通过Socket来读取数据，通过Sockt创建的DataStream能够从Socket中无限接收字符串，字符编码采用系统默认字符集。当Socket关闭时，Source停止读取。Socket提供了5个重载方法，但是有两个方法已经标记废弃。

socketTextStream(String hostname,int port)：指定Socket主机和端口，默认数据分隔符为换行符(\n)。
socketTextStream(String hostname,int port,String delimiter)：指定Socket主机和端口，数据分隔符为delimiter。
socketTextStream(String hostname,int port,String delimiter,long maxRetry)：该重载方法能够当与Socket断开时进行重连，重连次数由maxRetry决定，时间间隔为1秒。如果为0则表示立即终止不重连，如果为负数则表示一直重试。

DataStreamSource<String> socketDS = env.socketTextStream("localhost", 7777);

3.基于文件的预定义Source

基于文件创建DataStream主要有两种方式：readTextFile和readFile。(readFileStream已废弃)。readTextFile就是简单读取文件，而readFile的使用方式比较灵活。

readTextFile

readTextFile提供了两个重载方法：

readTextFile(String filePath)：逐行读取指定文件来创建DataStream，使用系统默认字符编码读取。
readTextFile(String filePath,String charsetName)：逐行读取文件来创建DataStream，使用charsetName编码读取。

DataSource<String> lineDS = env.readTextFile("D:\\test\\test.txt");
DataStreamSource<String> lineDS = env.readTextFile("D:\\test\\test.txt");

readFile

readFile通过指定的FileInputFormat来读取用户指定路径的文件。对于指定路径文件，我们可以使用不同的处理模式来处理，FileProcessingMode.PROCESS_ONCE模式只会处理文件数据一次，而FileProcessingMode.PROCESS_CONTINUOUSLY会监控数据源文件是否有新数据，如果有新数据则会继续处理。

readFile(FileInputFormat<T> inputFormat,
　　　　　　String filePath,
　　　　　　　　FileProcessingMode watchType,
　　　　　　　　　　long interval,TypeInformation typrInfo)

参数	说明	实例
inputFormat	创建DataStream指定的输入格式
filePath	读取的文件路径，为URI格式。既可以读取普通文件，可以读取HDFS文件	file:///some/local/file 或hdfs://host:port/file/path
watchType	文件数据处理方式	FileProcessingMode.PROCESS_ONCE或FileProcessingMode.PROCESS_CONTINUOUSLY
interval	在周期性监控Source的模式下(PROCESS_CONTINUOUSLY)，指定每次扫描的时间间隔	10
typeInformation	返回数据流的类型

readFile提供了几个便于使用的重载方法，但它们最终都是调用上面这个方法的。

readFile(FileInputFormat<T> inputFormat,String filePath)：处理方式默认使用FileProcessingMode.PROCESS_ONCE。
readFile(FileInputFormat<T> inputFormat,String filePath,FileProcessingMode watchType,long interval)：返回类型默认为inputFormat类型。

需要注意：在使用FileProcessingMode.PROCESS_CONTINUOUSLY时，当修改读取文件时，Flink会将文件整体内容重新处理，也就是打破了"exactly-once"。