数据转换（Transformation），即通过从一个或多个 DataStream 生成新的DataStream 的过程，是主要的数据处理的手段。Flink 提供了多种数据转换操作，基本可以满足所有的日常使用场景。

所有的转换操作可分为：

单 Single-DataStream

多 Multi-DataStream

物理分区

三种类型

今天我们讨论 单 DataStream 转换 的场景

Map

Map 操作对 DataStream[ T ] 数据进行处理，常用作对数据集的清洗和转换。

如下代码展示了，把 每个tuple 的第二个元素加1的过程

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.fromElements(("1",5),("2",6),("3",9)).map(m => (m._1,m._2 + 1))

除了可以直接在 map 方法中直接传入计算表达式，还可以使用 MapFunction 接口定义 map 函数逻辑，如下代码：

env.fromElements(("1",5),("2",6),("3",9)).map(new MapFunction[(String,Int),(String,Int)] {override def map(value: (String, Int)): (String, Int) = {(value._1,value._2 + 2)}})

FlatMap

FlatMap 算子主要应用处理输入一个元素，产生一个或者多个元素的计算场景，常见的例子就是 WordCount 中，将每一行的文本数据切割，生成单词序列。

env.fromElements("123 sffs ffe").flatMap(str => str.split(" "))

输入结果集是一个字符串：123 sffs ffe

输出结果集是三个字符串：123 和 sffs 和 ffe

Filter

该算子是按照条件对结果集进行筛选操作，将符合条件的数据集输出，将不符合条件的结果过滤掉。

如下代码筛选出所有偶数

env.fromCollection(List(1,2,3,4)).filter(_ % 2 == 0)

可以用在日志过滤，数据清洗等场景

KeyBy

根据指定的 key 将输入的 DataStream[ T ] 数据格式转换为 KeyedStream[ T ] ，相当于在数据集中执行 partition 操作，将相同的 key 值的数据放到相同的分区中，如下代码所示：

env.fromElements((1,5),(2,3),(4,5),(4,6)).keyBy(0).print()

根据第一个值来分区，结果是：

1> (4,5)

4> (2,3)

1> (4,6)

3> (1,5)

可以用在 对数据分组统计的场景

Reduce

该算子和 MapReduce 中的 reduce 的意思是一样的，主要是将 KeyedStream 的数据通过用户自定义的 ReduceFunction 滚动地进行数据聚合处理。

env.fromElements((1,5),(2,3),(4,5),(4,6)).keyBy(0).reduce((t1,t2) => (t1._1,t1._2 + t2._2)).print()

对第二个元素进行累加，结果如下：

3> (1,5)

4> (2,3)

1> (4,5)

1> (4,11)

Aggregations

这是一系列算子的统称，其实就是将 Reduce 中的算子进行了封装，封装的聚合操作有 sum，min，max，maxBy 等，这样用户就不同自己定义函数了

val env = StreamExecutionEnvironment.getExecutionEnvironment
env.fromElements((1,5),(2,3),(4,5),(4,6)).keyBy(0).sum(1).print()

env.fromElements((1,5),(2,3),(4,5),(4,6)).keyBy(0).min(1).print()

关注公众号，获得文章源码