《从0到1学习Flink》—— Data Source 介绍

前言

Data Sources 是什么呢？就字面意思其实就能够知道：数据来源。

Flink 作为一款流式计算框架，它可用来作批处理，即处理静态的数据集、历史的数据集；也能够用来作流处理，即实时的处理些实时数据流，实时的产生数据流结果，只要数据源源不断的过来，Flink 就可以一直计算下去，这个 Data Sources 就是数据的来源地。

Flink 中你可使用 StreamExecutionEnvironment.addSource(sourceFunction) 来为你的程序添加数据来源。

Flink 已经提供了若干实现好了的 source functions，固然你也能够经过实现 SourceFunction 来自定义非并行的 source 或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source，

Flink

StreamExecutionEnvironment 中可使用如下几个已实现的 stream sources，

总的来讲能够分为下面几大类：

基于集合

一、fromCollection(Collection) - 从 Java 的 Java.util.Collection 建立数据流。集合中的全部元素类型必须相同。

二、fromCollection(Iterator, Class) - 从一个迭代器中建立数据流。Class 指定了该迭代器返回元素的类型。

三、fromElements(T …) - 从给定的对象序列中建立数据流。全部对象类型必须相同。

1 2 3 4 5 6 7 8

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Event> input = env.fromElements( new Event(1, "barfoo", 1.0), new Event(2, "start", 2.0), new Event(3, "foobar", 3.0), ... );

四、fromParallelCollection(SplittableIterator, Class) - 从一个迭代器中建立并行数据流。Class 指定了该迭代器返回元素的类型。

五、generateSequence(from, to) - 建立一个生成指定区间范围内的数字序列的并行数据流。

基于文件

一、readTextFile(path) - 读取文本文件，即符合 TextInputFormat 规范的文件，并将其做为字符串返回。

1 2 3

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<String> text = env.readTextFile("file:///path/to/file");

二、readFile(fileInputFormat, path) - 根据指定的文件输入格式读取文件（一次）。

三、readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo) - 这是上面两个方法内部调用的方法。它根据给定的 fileInputFormat 和读取路径读取文件。根据提供的 watchType，这个 source 能够按期（每隔 interval 毫秒）监测给定路径的新数据（FileProcessingMode.PROCESS_CONTINUOUSLY），或者处理一次路径对应文件的数据并退出（FileProcessingMode.PROCESS_ONCE）。你能够经过 pathFilter 进一步排除掉须要处理的文件。

1 2 3 4 5

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<MyEvent> stream = env.readFile( myFormat, myFilePath, FileProcessingMode.PROCESS_CONTINUOUSLY, 100, FilePathFilter.createDefaultFilter(), typeInfo);

实现:

在具体实现上，Flink 把文件读取过程分为两个子任务，即目录监控和数据读取。每一个子任务都由单独的实体实现。目录监控由单个非并行（并行度为1）的任务执行，而数据读取由并行运行的多个任务执行。后者的并行性等于做业的并行性。单个目录监控任务的做用是扫描目录（根据 watchType 按期扫描或仅扫描一次），查找要处理的文件并把文件分割成切分片（splits），而后将这些切分片分配给下游 reader。reader 负责读取数据。每一个切分片只能由一个 reader 读取，但一个 reader 能够逐个读取多个切分片。

重要注意：

若是 watchType 设置为 FileProcessingMode.PROCESS_CONTINUOUSLY，则当文件被修改时，其内容将被从新处理。这会打破“exactly-once”语义，由于在文件末尾附加数据将致使其全部内容被从新处理。

若是 watchType 设置为 FileProcessingMode.PROCESS_ONCE，则 source 仅扫描路径一次而后退出，而不等待 reader 完成文件内容的读取。固然 reader 会继续阅读，直到读取全部的文件内容。关闭 source 后就不会再有检查点。这可能致使节点故障后的恢复速度较慢，由于该做业将从最后一个检查点恢复读取。

基于 Socket：

socketTextStream(String hostname, int port) - 从 socket 读取。元素能够用分隔符切分。

1 2 3 4 5 6 7 8

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<String, Integer>> dataStream = env .socketTextStream("localhost", 9999) // 监听 localhost 的 9999 端口过来的数据 .flatMap(new Splitter()) .keyBy(0) .timeWindow(Time.seconds(5)) .sum(1);

这个在 《从0到1学习Flink》—— Mac 上搭建 Flink 1.6.0 环境并构建运行简单程序入门 文章里用的就是基于 Socket 的 Word Count 程序。

自定义：

addSource - 添加一个新的 source function。例如，你能够 addSource(new FlinkKafkaConsumer011<>(…)) 以从 Apache Kafka 读取数据

说下上面几种的特色吧：

一、基于集合：有界数据集，更偏向于本地测试用

二、基于文件：适合监听文件修改并读取其内容

三、基于 Socket：监听主机的 host port，从 Socket 中获取数据

四、自定义 addSource：大多数的场景数据都是无界的，会源源不断的过来。好比去消费 Kafka 某个 topic 上的数据，这时候就须要用到这个 addSource，可能由于用的比较多的缘由吧，Flink 直接提供了 FlinkKafkaConsumer011 等类可供你直接使用。你能够去看看 FlinkKafkaConsumerBase 这个基础类，它是 Flink Kafka 消费的最根本的类。

1 2 3 4 5 6 7 8 9

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<KafkaEvent> input = env .addSource( new FlinkKafkaConsumer011<>( parameterTool.getRequired("input-topic"), //从参数中获取传进来的 topic new KafkaEventSchema(), parameterTool.getProperties()) .assignTimestampsAndWatermarks(new CustomWatermarkExtractor()));

Flink 目前支持以下图里面常见的 Source：

若是你想本身自定义本身的 Source 呢？

那么你就须要去了解一下 SourceFunction 接口了，它是全部 stream source 的根接口，它继承自一个标记接口（空接口）Function。

SourceFunction 定义了两个接口方法：

一、run ： 启动一个 source，即对接一个外部数据源而后 emit 元素造成 stream（大部分状况下会经过在该方法里运行一个 while 循环的形式来产生 stream）。

二、cancel ： 取消一个 source，也即将 run 中的循环 emit 元素的行为终止。

正常状况下，一个 SourceFunction 实现这两个接口方法就能够了。其实这两个接口方法也固定了一种实现模板。

好比，实现一个 XXXSourceFunction，那么大体的模板是这样的：(直接拿 FLink 源码的实例给你看看)

最后

本文主要讲了下 Flink 的常见 Source 有哪些而且简单的提了下如何自定义 Source。