这篇文章主要介绍“Flink流计算常用算子是什么”，在日常操作中，相信很多人在Flink流计算常用算子是什么问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”Flink流计算常用算子是什么”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！
Flink和Spark类似，也是一种一站式处理的框架；既可以进行批处理（DataSet），也可以进行实时处理（DataStream）。
所以下面将Flink的算子分为两大类：一类是DataSet，一类是DataStream。
DataSet 一、Source算子 1. fromCollection fromCollection：从本地集合读取数据
例：
val env = ExecutionEnvironment.getExecutionEnvironment
2. readTextFile readTextFile：从文件中读取：
val textDataSet: DataSet[String] = env.readTextFile("/data/a.txt") 3. readTextFile：遍历目录 readTextFile可以对一个文件目录内的所有文件，包括所有子目录中的所有文件的遍历访问方式：
val parameters = new Configuration
4. readTextFile：读取压缩文件 对于以下压缩类型，不需要指定任何额外的inputformat方法，flink可以自动识别并且解压。但是，压缩文件可能不会并行读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。
压缩方法 文件扩展名 是否可并行读取 DEFLATE .deflate no GZip .gz .gzip no Bzip2 .bz2 no XZ .xz no
val file = env.readTextFile("/data/file.gz") 二、Transform转换算子 因为Transform算子基于Source算子操作，所以首先构建Flink执行环境及Source算子，后续Transform算子操作基于此：
val env = ExecutionEnvironment.getExecutionEnvironment
1. map 将DataSet中的每一个元素转换为另外一个元素：
// 使用map将List转换为一个Scala的样例类
2. flatMap 将DataSet中的每一个元素转换为0...n个元素：
// 使用flatMap操作，将集合中的数据： 3. mapPartition 将一个分区中的元素转换为另一个元素：
// 使用mapPartition操作，将List转换为一个scala的样例类
4. filter 过滤出来一些符合条件的元素，返回boolean值为true 的元素：
val source: DataSet[String] = env.fromElements("java", "scala", "java")
5. reduce 可以对一个dataset或者一个group来进行聚合计算，最终聚合成一个元素：
// 使用 fromElements 构建数据源
6. reduceGroup 将一个dataset或者一个group聚合成一个或多个元素 。
// 使用 fromElements 构建数据源
7. minBy和maxBy 选择具有最小值或最大值的元素：
// 使用minBy操作，求List中每个人的最小值 8. Aggregate 在数据集上进行聚合求最值 （最大值、最小值）：
val data = new mutable.MutableList[(Int, String, Double)] Aggregate只能作用于元组上
注意：groupBy(0) ，否则会报一下错误:
9. distinct 去除重复的数据：
// 数据源使用上一题的
10. first 取前N个数：
input.first(2) // 取前两个数 11. join 将两个DataSet按照一定条件连接到一起，形成新的DataSet：
// s1 和 s2 数据集格式如下：
12. leftOuterJoin 左外连接,左边的Dataset中的每一个元素，去连接右边的元素
此外还有：
rightOuterJoin：右外连接,左边的Dataset中的每一个元素，去连接左边的元素
fullOuterJoin：全外连接,左右两边的元素，全部连接
下面以 leftOuterJoin 进行示例：
val data1 = ListBuffer[Tuple2[Int,String]]()
13. cross 交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建一个新的数据集
和join类似，但是这种交叉操作会产生笛卡尔积，在数据比较大的时候，是非常消耗内存的操作：
val cross = input1.cross(input2){
14. union 联合操作，创建包含来自该数据集和其他数据集的元素的新数据集,不会去重：
val unionData: DataSet[String] = elements1.union(elements2).union(elements3)
15. rebalance Flink也有数据倾斜的时候，比如当前有数据量大概10亿条数据需要处理，在处理过程中可能会发生如图所示的状况：
这个时候本来总体数据量只需要10分钟解决的问题，出现了数据倾斜，机器1上的任务需要4个小时才能完成，那么其他3台机器执行完毕也要等待机器1执行完毕后才算整体将任务完成；所以在实际的工作中，出现这种情况比较好的解决方案就是接下来要介绍的—rebalance （内部使用round robin方法将数据均匀打散。这对于数据倾斜时是很好的选择。）
// 使用rebalance操作，避免数据倾斜
16. partitionByHash 按照指定的key进行hash分区：
val data = new mutable.MutableList[(Int, Long, String)]
17. partitionByRange 根据指定的key对数据集进行范围分区：
val data = new mutable.MutableList[(Int, Long, String)]
18. sortPartition 根据指定的字段值进行分区的排序：
val data = new mutable.MutableList[(Int, Long, String)]
三、Sink算子 1. collect 将数据输出到本地集合：
result.collect() 2. writeAsText 将数据输出到文件
Flink支持多种存储设备上的文件，包括本地文件，hdfs文件等
Flink支持多种文件的存储格式，包括text文件，CSV文件等
// 将数据写入本地文件
DataStream 和DataSet一样，DataStream也包括一系列的Transformation操作。
一、Source算子 Flink可以使用 StreamExecutionEnvironment.addSource(source) 来为我们的程序添加数据来源。
Flink在流处理上的source和在批处理上的source基本一致。大致有4大类：
基于
本地集合 的source（Collection-based-source） 基于
文件 的source（File-based-source）- 读取文本文件，即符合 TextInputFormat 规范的文件，并将其作为字符串返回 基于
网络套接字 的source（Socket-based-source）- 从 socket 读取。元素可以用分隔符切分。 自定义 的source（Custom-source）下面使用addSource将Kafka数据写入Flink为例：
如果需要外部数据源对接，可使用addSource，如将Kafka数据写入Flink， 先引入依赖：
org.apache.flink flink-connector-kafka-0.11_2.11 1.10.0 将Kafka数据写入Flink：
val properties = new Properties()基于网络套接字的：
val source = env.socketTextStream("IP", PORT) 二、Transform转换算子 1. map 将DataSet中的每一个元素转换为另外一个元素：
dataStream.map { x => x * 2 } 2. FlatMap 采用一个数据元并生成零个，一个或多个数据元。将句子分割为单词的flatmap函数：
dataStream.flatMap { str => str.split(" ") } 3. Filter 计算每个数据元的布尔函数，并保存函数返回true的数据元。过滤掉零值的过滤器：
dataStream.filter { _ != 0 } 4. KeyBy 逻辑上将流分区为不相交的分区。具有相同Keys的所有记录都分配给同一分区。在内部，keyBy（）是使用散列分区实现的。指定键有不同的方法。
此转换返回KeyedStream，其中包括使用被Keys化状态所需的KeyedStream：
dataStream.keyBy(0) 5. Reduce 被Keys化数据流上的“滚动”Reduce。将当前数据元与最后一个Reduce的值组合并发出新值：
keyedStream.reduce { _ + _ } 6. Fold 具有初始值的被Keys化数据流上的“滚动”折叠。将当前数据元与最后折叠的值组合并发出新值：
val result: DataStream[String] = keyedStream.fold("start")((str, i) => { str + "-" + i })
7. Aggregations 在被Keys化数据流上滚动聚合。min和minBy之间的差异是min返回最小值，而minBy返回该字段中具有最小值的数据元（max和maxBy相同）：
keyedStream.sum(0);
8. Window 可以在已经分区的KeyedStream上定义Windows。Windows根据某些特征（例如，在最后5秒内到达的数据）对每个Keys中的数据进行分组。这里不再对窗口进行详解，有关窗口的完整说明，请查看这篇文章：Flink 中极其重要的 Time 与 Window 详细解析
dataStream.keyBy(0).window(TumblingEventTimeWindows.of(Time.seconds(5))); 9. WindowAll Windows可以在常规DataStream上定义。Windows根据某些特征（例如，在最后5秒内到达的数据）对所有流事件进行分组。
注意：在许多情况下，这是非并行转换。所有记录将收集在windowAll 算子的一个任务中。
dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5))) 10. Window Apply 将一般函数应用于整个窗口。
注意：如果您正在使用windowAll转换，则需要使用AllWindowFunction。
下面是一个手动求和窗口数据元的函数：
windowedStream.apply { WindowFunction }
11. Window Reduce 将函数缩减函数应用于窗口并返回缩小的值：
windowedStream.reduce { _ + _ } 12. Window Fold 将函数折叠函数应用于窗口并返回折叠值：
val result: DataStream[String] = windowedStream.fold("start", (str, i) => { str + "-" + i })
13. Union 两个或多个数据流的联合，创建包含来自所有流的所有数据元的新流。注意：如果将数据流与自身联合，则会在结果流中获取两次数据元：
dataStream.union(otherStream1, otherStream2, ...) 14. Window Join 在给定Keys和公共窗口上连接两个数据流：
dataStream.join(otherStream)).equalTo()
15. Interval Join 在给定的时间间隔内使用公共Keys关联两个被Key化的数据流的两个数据元e1和e2，以便e1.timestamp + lowerBound <= e2.timestamp <= e1.timestamp + upperBound
am.intervalJoin(otherKeyedStream)
16. Window CoGroup 在给定Keys和公共窗口上对两个数据流进行Cogroup：
dataStream.coGroup(otherStream)
17. Connect “连接”两个保存其类型的数据流。连接允许两个流之间的共享状态：
DataStream someStream = ... DataStream otherStream = ... ConnectedStreams connectedStreams = someStream.connect(otherStream)
18. CoMap，CoFlatMap 类似于连接数据流上的map和flatMap：
connectedStreams.map(
19. Split 根据某些标准将流拆分为两个或更多个流：
val split = someDataStream.split(
20. Select 从拆分流中选择一个或多个流：
SplitStream split;DataStream even = split.select("even");DataStream odd = split.select("odd");DataStream all = split.select("even","odd")
三、Sink算子 支持将数据输出到：
除此之外，还支持：
下面以sink到kafka为例：
val sinkTopic = "test"到此，关于“Flink流计算常用算子是什么”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注创新互联网站，小编会继续努力为大家带来更多实用的文章！
