彻底明白Flink系统学习14：【Flink1.7】DataSet 编程之Transformations详解

问题导读
1.DataSet与 Datastream Transformations 都有哪些API？
2.DataSet MapPartition的作用是什么？
3.ReduceGroup与Reduce有什么区别？
4.Join Hints是什么？
5.Flink是否所有外部联接类型都支持所有执行策略？
6.CoGroup与join的区别是什么？
7.Cross使用需要注意什么？
8.Range-Partition可以解决什么问题？

上一篇：
彻底明白Flink系统学习13：【Flink1.7】流连接器介绍及如何添加连接器
http://www.aboutyun.com/forum.php?mod=viewthread&tid=26512

相关篇：
彻底明白Flink系统学习9：【Flink1.7编程】数据流Transformations介绍2窗口及相关操作
http://www.aboutyun.com/forum.php?mod=viewthread&tid=26469

尽管许多人都很欣赏大多数行业中流数据处理的潜在价值，但是有许多使用情况，人们认为没有必要以流方式处理数据。在所有这些情况下，批处理都是可行的方法。到目前为止，Hadoop一直是数据处理的默认选择。但是，Flink还支持DataSet API的批处理数据处理。

对于Flink，批处理是流处理的特例。在本文中，我们将查看有关DataSet API的详细信息。

Transformations将数据集从一种形式转换为另一种形式。输入可以是一个或多个数据集，输出也可以是零，或者一个或多个数据流。也就是说Transformations将一个或多个DataSet转换为新的DataSet。程序可以将多个Transformations组合到复杂的程序集中。

Map
取每个元素转换为另外的元素
[mw_shl_code=scala,true]data.map { x => x.toInt }
[/mw_shl_code]

FlatMap
取每个元素转换为0, 1, 或则更多元素
[mw_shl_code=scala,true]data.flatMap { str => str.split(" ") }
[/mw_shl_code]

MapPartition
单个函数调用并行分区，该函数将分区作为“迭代器”，可以产生任意数量的结果。每个分区中的元素数量取决于并行度和以前的operations。
MapPartition和map是类似的，不过MapPartition相对来说效率高一些。
map有多少个元素，map就会被调用多少次
MapPartition对于数据有多少分区，就会被调用多少次。
[mw_shl_code=scala,true]data.mapPartition { in => in map { (_, 1) } }
[/mw_shl_code]

Filter
计算每个元素的布尔函数，并保留函数返回true的元素。
重要信息：系统假定该函数不会修改元素。违反此假设可能会导致错误的结果。

[mw_shl_code=scala,true]data.filter { _ > 1000 }
[/mw_shl_code]

Reduce
通过将两个元素重复组合成一个元素，将一组元素组合成一个元素。 Reduce可以应用于完整数据集或分组数据集。
[mw_shl_code=scala,true]data.reduce { _ + _ }
[/mw_shl_code]

ReduceGroup
将一组元素组合成一个或多个元素。 ReduceGroup可以应用于完整数据集或分组数据集。
[mw_shl_code=scala,true]data.reduceGroup { elements => elements.sum }
[/mw_shl_code]

Aggregate
将一组值聚合为单个值。聚合函数可以被认为是内置的reduce函数。聚合可以应用于完整数据集或分组数据集。
[mw_shl_code=scala,true]val input: DataSet[(Int, String, Double)] = // [...]
val output: DataSet[(Int, String, Double)] = input.aggregate(SUM, 0).aggregate(MIN, 2)[/mw_shl_code]

可以使用简写语法进行最小，最大和求和聚合。
[mw_shl_code=scala,true]val input: DataSet[(Int, String, Double)] = // [...]
val output: DataSet[(Int, String, Double)] = input.sum(0).min(2)[/mw_shl_code]

Distinct
返回数据集的不同元素。它相对于元素的所有字段或字段子集从输入DataSet中删除重复条目。
[mw_shl_code=scala,true]data.distinct()[/mw_shl_code]

Join
通过创建在其key上相等的所有元素对来连接两个数据集。可以使用JoinFunction将元素对转换为单个元素，或使用FlatJoinFunction将元素对转换为任意多个（包括无）元素。
补充：如何定义key:以DataSet分组来说：
[mw_shl_code=scala,true]DataSet<...> input = // [...]
DataSet<...> reduced = input
.groupBy(/*define key here*/)
.reduceGroup(/*do something*/);[/mw_shl_code]
我们看到groupBy中可以自定义key

下面情况下，元组字段用作键。
“0”是第一个元组的连接字段
“1”是第二个元组的连接字段。
[mw_shl_code=scala,true]// In this case tuple fields are used as keys. "0" is the join field on the first tuple
// "1" is the join field on the second tuple.
val result = input1.join(input2).where(0).equalTo(1)[/mw_shl_code]

可以通过Join Hints来指定运行时执行join的方式。这些提示描述了连接是通过分区还是广播，以及它是使用基于排序的（sort-based）还是基于散列(hash-based)的算法。如果没有指定链接方式，系统将尝试对输入大小进行评估，并根据这些评估选择最佳策略。
注意，连接transformation 仅适用于等连接。其他连接类型需要使用OuterJoin或CoGroup表示。

补充：
上面我们看到Join Hints翻译为中文是链接提示的意思。那么它的具体是什么含义，我们慢慢来看：
Flink运行时可以以各种方式执行连接。在不同情况下，每种可能的方式都优于其他方式。系统会尝试自动选择合理的方式，但允许手动选择策略，强制执行外连接的特定方式。对于那么特定的方式有哪些？我们这里举下面例子：
[mw_shl_code=scala,true]val input1: DataSet[SomeType] = // [...]
val input2: DataSet[AnotherType] = // [...]
// hint that the second DataSet is very small
val result1 = input1.leftOuterJoin(input2, JoinHint.REPARTITION_SORT_MERGE).where("id").equalTo("key")
val result2 = input1.rightOuterJoin(input2, JoinHint.BROADCAST_HASH_FIRST).where("id").equalTo("key")[/mw_shl_code]
我们可以看到JoinHint.BROADCAST_HASH_FIRST，这里或许我们对JoinHint有了认识，我们接着往下看。

以下hints 可用：
OPTIMIZER_CHOOSES：相当于不提供任何提示，将选择留给系统。
BROADCAST_HASH_FIRST：广播第一个输入并从中构建一个哈希表，由第二个输入探测。如果第一个输入非常小，那么这是一个很好的策略
BROADCAST_HASH_SECOND：广播第二个输入并从中构建一个哈希表，由第一个输入探测。如果第二个输入非常小，这是一个很好的策略。
REPARTITION_HASH_FIRST：系统对每个输入进行分区（shuffle）（除非输入已经分区）并从第一个输入构建哈希表。如果第一个输入小于第二个输入，则此策略很好，尽管两个输入仍然很大。
REPARTITION_HASH_SECOND：系统对每个输入进行分区（shuffle）（除非输入已经分区）并从第二个输入构建哈希表。如果第二个输入小于第一个输入，则此策略很好，尽管两个输入仍然很大。
REPARTITION_SORT_MERGE：系统对每个输入进行分区（shuffle）（除非输入已经被分区）并对每个输入进行排序（除非它已经排序）。输入通过已排序输入的流合并来连接。如果已经对一个或两个输入进行了排序，则此策略很好。
注意：并非所有外部联接类型都支持所有执行策略。

LeftOuterJoin supports:
- OPTIMIZER_CHOOSES
- BROADCAST_HASH_SECOND
- REPARTITION_HASH_SECOND
- REPARTITION_SORT_MERGE
RightOuterJoin supports:
- OPTIMIZER_CHOOSES
- BROADCAST_HASH_FIRST
- REPARTITION_HASH_FIRST
- REPARTITION_SORT_MERGE
FullOuterJoin supports:
- OPTIMIZER_CHOOSES
- REPARTITION_SORT_MERGE

OuterJoin
对两个数据集执行左、右或完全外部连接。外部联接类似于常规（内部）联接，在键上创建相等的所有元素对。此外，如果在另一侧没有找到匹配的键，则保留“外部”的记录（左、右，或者在满的情况下两者都保留）。将匹配元素对（或一个元素和另一个输入的“空”值）赋给JoinFunction以将元素对转换为单个元素，或给FlatJoinFunction以将元素对转换为任意多个（包括无）元素。
[mw_shl_code=scala,true]val joined = left.leftOuterJoin(right).where(0).equalTo(1) {
(left, right) =>
   val a = if (left == null) "none" else left._1
   (a, right)
  }[/mw_shl_code]

CoGroup:

reduce操作的二维变体。将一个或多个字段上的每个输入分组，然后join组。每对组调用transformation 函数。其实，这个跟join其实是类似的，除了输出匹配的元素对以外，未能匹配的元素也会输出，而join不同，输出则不同。

[mw_shl_code=scala,true]data1.coGroup(data2).where(0).equalTo(1)
[/mw_shl_code]
关于CoGroup等，推荐文章
Flink中cogroup, join和coflatmap比较
http://www.aboutyun.com/forum.php?mod=viewthread&tid=26532

Cross
构建两个输入的笛卡尔积（交叉乘积），创建所有元素对。可选择使用CrossFunction将元素对转换为单个元素
[mw_shl_code=scala,true]val data1: DataSet[Int] = // [...]
val data2: DataSet[String] = // [...]
val result: DataSet[(Int, String)] = data1.cross(data2)[/mw_shl_code]
注意：Cross可能是一个计算密集型操作，甚至可以挑战大型集群！建议使用crossWithTiny（）和crossWithHuge（）来提示系统DataSet大小。

Union
生成两个数据集的并集。
[mw_shl_code=scala,true]data.union(data2)
[/mw_shl_code]

Rebalance
均匀地重新平衡数据集的并行分区以消除数据偏差。只有类似Map的转换可以跟着rebalance操作使用。
[mw_shl_code=scala,true]val data1: DataSet[Int] = // [...]
val result: DataSet[(Int, String)] = data1.rebalance().map(...)[/mw_shl_code]

Hash-Partition
根据给定的key对一个数据集进行hash分区。Key可以是position keys, expression keys, 或者key selector functions
[mw_shl_code=scala,true]val in: DataSet[(Int, String)] = // [...]
val result = in.partitionByHash(0).mapPartition { ... }[/mw_shl_code]

Range-Partition
根据给定的key对一个数据集进行范围分区。Key可以是position keys, expression keys, 或者key selector functions
[mw_shl_code=scala,true]val in: DataSet[(Int, String)] = // [...]
val result = in.partitionByRange(0).mapPartition { ... }[/mw_shl_code]

补充：
Hash-Partition与Range-Partition的区别？
HashPartitioner分区的原理，对于给定的key，计算其hashCode，并除于分区的个数取余，如果余数小于0，则用余数+分区的个数，最后返回的值就是这个key所属的分区ID。
RangePartitioner分区器的主要作用就是将一定范围内的数映射到某一个分区内，分界算法对应的函数是rangeBounds。RangePartitioner分区的优势，可以解决HashPartitioner中可能出现的数据不均匀现象，每个分区的数据，都比另外的分区的数据大或则小。

Custom Partitioning

使用自定义Partitioner函数分配基于key的指定分区的记录，也就是说根据我们的自定义分区，来对数据进行划分。这个有点像我们的Hbase预分区，而这里我们只是指定了或则自定义了分区器。
注意：此方法仅适用于单个字段。
[mw_shl_code=scala,true]val in: DataSet[(Int, String)] = // [...]
val result = in
  .partitionCustom(partitioner, key).mapPartition { ... }[/mw_shl_code]

Sort Partition
以指定的顺序在指定的字段上对本地数据集的所有分区排序。 Fields 可以指定为 tuple positions 或则字段expressions.。通过sortPartition()调用来完成对多个字段的排序。
[mw_shl_code=scala,true]val in: DataSet[(Int, String)] = // [...]
val result = in.sortPartition(1, Order.ASCENDING).mapPartition { ... }[/mw_shl_code]

First-n

返回数据集的前n个（任意）元素。 First-n可以应用于常规数据集，分组数据集或分组排序数据集。可以将分组keys 指定为 key-selector函数，uple positions 或则class fields。
[mw_shl_code=scala,true]val in: DataSet[(Int, String)] = // [...]
// regular data set
val result1 = in.first(3)
// grouped data set
val result2 = in.groupBy(0).first(3)
// grouped-sorted data set
val result3 = in.groupBy(0).sortGroup(1, Order.ASCENDING).first(3)[/mw_shl_code]

以下transformations 可用于元组的数据集：

MinBy / MaxBy
从一组元组中选择一个元组，其元组的一个或多个字段的值最小（最大）。用于比较的字段必须是有效的key字段，即可比较的字段。如果多个元组具有最小（最大）字段值，则返回这些元组的任意元组。 MinBy（MaxBy）可以应用于完整数据集或分组数据集。
[mw_shl_code=scala,true]val in: DataSet[(Int, Double, String)] = // [...]
// a data set with a single tuple with minimum values for the Int and String fields.
val out: DataSet[(Int, Double, String)] = in.minBy(0, 2)
// a data set with one tuple for each group with the minimum value for the Double field.
val out2: DataSet[(Int, Double, String)] = in.groupBy(2)
                                          .minBy(1)[/mw_shl_code]

Project
从元组中选择字段的子集【Java】
[mw_shl_code=java,true]DataSet<Tuple3<Integer, Double, String>> in = // [...]
DataSet<Tuple2<String, Integer>> out = in.project(2,0);[/mw_shl_code]

关注最新经典文章，欢迎关注公众号