DataStream流处理算子
Source算子（数据读入）
- Flink可以使用StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源
基于本地集合的source
DataStream<String> words = env.fromElements("hello", "flink", "stream");
基于文件的source
readTextFile(path)
- 读取指定路径的文本文件
readFile(fileInputFormat, path)
- 按指定的文件输入格式（fileInputFormat）读取指定路径path的文件
readFile(fileInputFormat, path, watchType, interval, pathFilter)
- watchType：定义了文件源的行为，可以是FileProcessingMode.PROCESS_CONTINUOUSLY，定期监听路径下的新数据；或者是FileProcessingMode.PROCESS_ONCE，只处理一次当前在路径中的数据并退出
- interval：当watchType设置为PROCESS_CONTINUOUSLY时，这个参数定义了文件系统被监控的间隔时间（毫秒）
- pathFilter：可选的PathFilter实例，用于进一步过滤要处理的文件
DataStream<String> text = env.readTextFile(inputPath);
基于网络套接字的source
- socketTextStream(host, port): 从指定的主机和端口读取文本行，返回一个字符串数据流
DataStream<String> text = env.socketTextStream("hadoop102", 7777);
从Hadoop兼容的文件系统读取
- readTextFile(FileSystemPath path): 从Hadoop文件系统读取文本文件
DataStream<String> text = env.readTextFile(new Path("hdfs:///path/to/your/file.txt"));
自定义source
- addSource(new RichParallelSourceFunction()): 实现自定义的Source功能
DataStream<String> customSource = env.addSource(new RichParallelSourceFunction<String>() {
@Override
public void run(SourceContext<String> ctx) throws Exception {
// 自定义数据生成逻辑
}
});
Transform转换算子（数据处理）
| 转换算子 | 描述 |
|---|---|
| map | 将DataSet中的每一个元素转换为另外一个元素 |
| FlatMap | 采用一个数据元并生成零个，一个或多个数据元，将句子分割为单词的flatmap函数 |
| FlatMap | 采用一个数据元并生成零个，一个或多个数据元 |
| Filter | 计算每个数据元的布尔函数，并保存函数返回true的数据元 |
| KeyBy | 逻辑上将流分区为不相交的分区。具有相同Keys的所有记录都分配给同一分区 |
| Reduce | 将当前数据元与最后一个Reduce的值组合并发出新值 |
| Fold | 将当前数据元与最后折叠的值组合并发出新值 |
| Aggregations | 在被Keys化数据流上滚动聚合，min和minBy之间的差异是min返回最小值，而minBy返回该字段中具有最小值的数据元 |
| Window | 可以在已经分区的KeyedStream上定义Windows |
| Union | 两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流 |
| connect | “连接”两个保存其类型的数据流 |
| Split | 根据某些标准将流拆分为两个或更多个流 |
| Select | 从拆分流中选择一个或多个流 |
dataStream.map { x => x * 2 }
//将字符串按空格分割
dataStream.flatMap { str => str.split(" ") }
//过滤零值
dataStream.filter { _ != 0 }
//按照数据流中每个元素的第一个字段进行分组，比如Tuple<t1,t2>，按照t1分组
dataStream.keyBy(0)
//对每个分组的数据流中的元素进行累加
keyedStream.reduce { _ + _ }
// 序列（1,2,3,4,5），输出结果“start-1”，“start-1-2”，“start-1-2-3”，...
val result: DataStream[String] = keyedStream.fold("start")((str, i) => { str + "-" + i })
keyedStream.sum(0);
keyedStream.min(0);
keyedStream.max(0);
keyedStream.minBy(0);
keyedStream.maxBy(0);
Sink算子（数据输出）
print()
- 将数据输出到标准输出（控制台）
dataStream.print();
writeAsText()
- 将数据以文本形式写入到文件系统中的文件，包括本地文件，HDFS文件
dataStream.writeAsText("output.txt");
writeAsCsv()
- 将数据以逗号分隔值（CSV）的格式写入到文件系统中的文件
dataStream.writeAsCsv("output.csv");
addSink(SinkFunction)
- 使用自定义的SinkFunction将数据输出到外部系统