1 概述
Flink中的DataStream程序是实现数据流转换的常规程序(例如,过滤,更新状态,定义窗口,聚合)。
最初从各种源(例如,消息队列,套接字流,文件)创建数据流。
结果通过接收器返回,接收器可以例如将数据写入文件或标准输出(例如命令行终端)。
Flink程序可以在各种环境中运行,独立运行或嵌入其他程序中。 执行可以在本地JVM中执行,也可以在许多计算机的集群上执行。
有关Flink API基本概念的介绍,请参阅
https://blog.csdn.net/qq_33589510/article/details/89893394
2 入门案例
以下程序是流窗口字数统计应用程序的完整工作示例,它在5秒窗口中对来自Web套接字的单词进行计数。 您可以复制并粘贴代码以在本地运行它。
import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.util.Collector; public class WindowWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); DataStream<Tuple2<String, Integer>> dataStream = env .socketTextStream("localhost", 9999) .flatMap(new Splitter()) .keyBy(0) .timeWindow(Time.seconds(5)) .sum(1); dataStream.print(); env.execute("Window WordCount"); } public static class Splitter implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String sentence, Collector<Tuple2<String, Integer>> out) throws Exception { for (String word: sentence.split(" ")) { out.collect(new Tuple2<String, Integer>(word, 1)); } } } }
nc -lk 9999
只需键入一些单词就可以返回一个新单词。 这些将是字数统计程序的输入。 如果要查看大于1的计数,请在5秒内反复键入相同的单词(如果不能快速输入,则将窗口大小从5秒增加☺)。
- Socket输入
- 程序输出
创建一个新数据流,其中包含从套接字无限接收的字符串。 接收的字符串由系统的默认字符集解码,使用“\ n”作为分隔符。 当socket关闭时,阅读器立即终止。
Scala版本
3 Data source
源是您的程序从中读取输入的位置。可以使用
StreamExecutionEnvironment.addSource(sourceFunction)
将源附加到程序
Flink附带了许多预置实现的源函数,但你可以通过为非并行源实现SourceFunction,或者通过实现ParallelSourceFunction接口或为并行源扩展RichParallelSourceFunction来编写自己的自定义源。
可以从StreamExecutionEnvironment访问几个预定义的流源:
3.1 基于文件
readTextFile(path)
TextInputFormat逐行读取文本文件,即符合规范的文件,并将它们作为字符串返回。
readFile(fileInputFormat, path)
按指定的文件输入格式指定读取(一次)文件。
readFile(fileInputFormat, path, watchType, interval, pathFilter, typeInfo)
这是前两个内部调用的方法。它path根据给定的内容读取文件fileInputFormat。根据提供的内容watchType,此源可以定期监视(每intervalms)新数据(FileProcessingMode.PROCESS_CONTINUOUSLY)的路径,或者处理当前在路径中的数据并退出(FileProcessingMode.PROCESS_ONCE)。使用该pathFilter,用户可以进一步排除正在处理的文件。
实现:
在引擎盖下,Flink将文件读取过程分为两个子任务
目录监控
数据读取
这些子任务中的每一个都由单独的实体实现。监视由单个非并行(并行性= 1)任务实现,而读取由并行运行的多个任务执行。
后者的并行性等于工作并行性。单个监视任务的作用是扫描目录(定期或仅一次,具体取决于watchType),找到要处理的文件,将它们分层分割,并将这些拆分分配给下游读卡器。读者是那些将阅读实际数据的人。每个分割仅由一个读取器读取,而读取器可以逐个读取多个分割。
1.如果watchType设置为FileProcessingMode.PROCESS_CONTINUOUSLY,则在修改文件时,将完全重新处理其内容。这可以打破“完全一次”的语义,因为在文件末尾追加数据将导致其所有内容被重新处理。
2.如果watchType设置为FileProcessingMode.PROCESS_ONCE,则源扫描路径一次并退出,而不等待读者完成读取文件内容。当然读者将继续阅读,直到读取所有文件内容。在该点之后关闭源将导致不再有检查点。这可能会导致节点发生故障后恢复速度变慢,因为作业将从上一个检查点恢复读取。
3.2 基于Socket
socketTextStream
从套接字读取。数据元可以用分隔符分隔。
3.3 基于集合
fromCollection(Collection)
从Java Java.util.Collection创建数据流。集合中的所有数据元必须属于同一类型。
fromCollection(Iterator, Class)
从迭代器创建数据流。该类指定迭代器返回的数据元的数据类型。
fromElements(T …)
从给定的对象序列创建数据流。所有对象必须属于同一类型。
fromParallelCollection(SplittableIterator, Class)
并行地从迭代器创建数据流。该类指定迭代器返回的数据元的数据类型。
generateSequence(from, to)
并行生成给定间隔中的数字序列。
3.4 自定义数据源方式SourceFunction
使用用户定义的源函数为任意源功能创建DataStream。
默认情况下,源具有1的并行性。
要启用并行执行,用户定义的源应
- 实现
ParallelSourceFunction
或继承RichParallelSourceFunction
在这些情况下,生成的源将具有环境的并行性。
要改变它,然后调用DataStreamSource.setParallelism(int)
- addSource
附加新的源函数。例如,要从Apache Kafka中读取,您可以使用 addSource(new FlinkKafkaConsumer08<>(…))