Flink常用流式转换算子

1.Map【DataStream -> DataStream】

调用MapFunction对DataStream进行处理，形成新的DataStream。一对一

val line: DataStream[String] = env.socketTextStream("node4", 9999)
val result: DataStream[String] = line.map(_ + "#")
result.print()

// 输入
hello java
hello spark  
// 输出
hello java#
hello spark#

2.FlatMap【DataStream -> DataStream]】

该算子主要应用处理输入一个元素产生一个或者多个 元素的计算场景，如经典案例WordCount。

val line: DataStream[String] = env.socketTextStream("node4", 9999)
val result: DataStream[String] = line.flatMap(_.split(" "))
result.print()

// 输入
hello java
hello spark  
// 输出
hello
java
hello
spark

3.Filter 【DataStream -> DataStream】

该算子按照条件对输入数据进行筛选操作，将符合条件的数据集输出，将不符合条件的数据过滤掉，类似于SQL中的where。

val line: DataStream[String] = env.socketTextStream("node4", 9999)
val filter: DataStream[String] = line.filter(_.endsWith("java"))
filter.print()

// 输入
hello java
hello spark  
// 输出
hello java

4.keyBy【DataStream -> KeyedStream】

该算子根据指定的Key将输入的DataStream数据格式转换为KeyedStream，相同Key的数据数据会分组放置到同一Slot中。

val value: DataStream[(String, String)] = env.fromElements(("湖北", "武汉"), ("湖南", "长沙"), ("湖北", "荆州"))
val keyBy: KeyedStream[(String, String), Tuple] = value.keyBy(0)
keyBy.print()

// 输出
2> (湖南,长沙)
4> (湖北,武汉)
4> (湖北,荆州)

5.reduce【-> DataStream】

将输入的KeyedStream通过传入的ReduceFnction滚动的进行数据实时聚合处理，其中定义的ReduceFunction必须满足运算结合律和交换律。

val source: DataStream[(String, String, Integer)] = env.fromElements(("湖北", "武汉", 1), ("湖南", "长沙", 1), ("湖北", "荆州", 1))
val keyBy: KeyedStream[(String, String, Integer), Tuple] = source.keyBy(0)
val reduceResult: DataStream[(String, String, Integer)] = keyBy.reduce((t1, t2) => {
  (t1._1, t1._2 + "&" + t2._2, t1._3 + t2._3)
})
reduceResult.print()

// 结果
4> (湖北,武汉,1)
4> (湖北,武汉&荆州,2)
2> (湖南,长沙,1)

6.Aggregations【KeyedStream -> DataStream】

Aggregations是KeyedStream接口提供的聚合算子，根据指定的字段进行实时聚合操作，滚动的产生一系列数据聚合结果。其实就是将Reduce算子中的函数进行了封装，封装的聚合操作有sum、min、minBy、max、maxBy等。

6.1 sum

将数据集中指定的字段进行累加。（注意打印结果湖北组地级市的值）

val source: DataStream[(String, String, Integer)] = env.fromElements(("湖北", "武汉", 2), ("湖南", "长沙", 1), ("湖北", "荆州", 1))
val keyBy: KeyedStream[(String, String, Integer), Tuple] = source.keyBy(0)
val value: DataStream[(String, String, Integer)] = keyBy.sum(2)
value.print()

// 打印结果
4> (湖北,武汉,2)
4> (湖北,武汉,3)
2> (湖南,长沙,1)

6.2 min和minBy

min：获取指定字段的最小值，但不是最小的那条记录。最终返回值是(湖北,武汉,1)

minBy:获取指定字段的最小值，同时也是最小值的那条记录。最终返回值是(湖北,荆州,1)

val source: DataStream[(String, String, Integer)] = env.fromElements(("湖北", "武汉", 2), ("湖南", "长沙", 1), ("湖北", "荆州", 1))
val keyBy: KeyedStream[(String, String, Integer), Tuple] = source.keyBy(0)
keyBy.min(2).print()
keyBy.minBy(2).print()

// min打印结果：
4> (湖北,武汉,2)
4> (湖北,武汉,1)
2> (湖南,长沙,1)
// minBy打印结果
4> (湖北,武汉,2)
4> (湖北,荆州,1)
2> (湖南,长沙,1)

6.3 max和maxBy

max和maxBy的区别类似

val source: DataStream[(String, String, Integer)] = env.fromElements(("湖北", "武汉", 2), ("湖南", "长沙", 1), ("湖北", "荆州", 1), ("湖南", "湘潭", 3))

val keyBy: KeyedStream[(String, String, Integer), Tuple] = source.keyBy(0)
keyBy.max(2).print()
keyBy.maxBy(2).print()

// max打印结果：
2> (湖南,长沙,1)
2> (湖南,长沙,3)
4> (湖北,武汉,2)
4> (湖北,武汉,2)
// maxBy打印结果
2> (湖南,长沙,1)
2> (湖南,湘潭,3)
4> (湖北,武汉,2)
4> (湖北,武汉,2)