文江博客开发文档 Flink 1.7 - SNAPSHOT 中文文档文章详情

文章来源于网络收集而来，版权归原创者所有，如有侵权请及时联系！

数据集转换

发布于 2025-05-02 18:19:14 字数 57578 浏览 0 评论 0 收藏

本文档深入介绍了 DataSet 上的可用转换。有关 Flink Java API 的一般介绍，请参阅编程指南。

有关在具有密集索引的数据集中压缩数据元，请参阅 Zip 数据元指南。

Map

Map 转换在 DataSet 的每个数据元上应用用户定义的 map 函数。它实现了一对一的映射，也就是说，函数必须返回一个数据元。

以下代码将 Integer 对的 DataSet 转换为 Integers 的 DataSet：

Java
Scala
Python

// MapFunction that adds two integer values
public class IntAdder implements MapFunction<Tuple2<Integer, Integer>, Integer> {
  @Override
  public Integer map(Tuple2<Integer, Integer> in) {
  return in.f0 + in.f1;
  }
}

// [...]
DataSet<Tuple2<Integer, Integer>> intPairs = // [...]
DataSet<Integer> intSums = intPairs.map(new IntAdder());

val intPairs: DataSet[(Int, Int)] = // [...] val intSums = intPairs.map { pair => pair._1 + pair._2 }

 intSums = intPairs.map(lambda x: sum(x))

FlatMap

FlatMap 转换在 DataSet 的每个数据元上应用用户定义的平面映射函数。map 函数的这种变体可以为每个输入数据元返回任意多个结果数据元（包括 none）。

以下代码将文本行的 DataSet 转换为单词的 DataSet：

Java
Scala
Python

// FlatMapFunction that tokenizes a String by whitespace characters and emits all String tokens.
public class Tokenizer implements FlatMapFunction<String, String> {
  @Override
  public void flatMap(String value, Collector<String> out) {
  for (String token : value.split("\\W")) {
    out.collect(token);
  }
  }
}

// [...]
DataSet<String> textLines = // [...]
DataSet<String> words = textLines.flatMap(new Tokenizer());

val textLines: DataSet[String] = // [...] val words = textLines.flatMap { _.split(" ") }

 words = lines.flat_map(lambda x,c: [line.split() for line in x])

MapPartition

MapPartition 在单个函数调用中转换并行分区。map-partition 函数将分区作为 Iterable 获取，并且可以生成任意数量的结果值。每个分区中的数据元数量取决于并行度和先前的算子操作。

以下代码将文本行的 DataSet 转换为每个分区的计数数据集：

Java
Scala
Python

public class PartitionCounter implements MapPartitionFunction<String, Long> {

  public void mapPartition(Iterable<String> values, Collector<Long> out) {
  long c = 0;
  for (String s : values) {
    c++;
  }
  out.collect(c);
  }
}

// [...]
DataSet<String> textLines = // [...]
DataSet<Long> counts = textLines.mapPartition(new PartitionCounter());

val textLines: DataSet[String] = // [...]
// Some is required because the return value must be a Collection.
// There is an implicit conversion from Option to a Collection. val counts = texLines.mapPartition { in => Some(in.size) }

 counts = lines.map_partition(lambda x,c: [sum(1 for _ in x)])

Filter

Filter 转换在 DataSet 的每个数据元上应用用户定义的过滤器函数，并仅保存函数返回的数据元 true 。

以下代码从 DataSet 中删除所有小于零的整数：

Java
Scala
Python

// FilterFunction that filters out all Integers smaller than zero.
public class NaturalNumberFilter implements FilterFunction<Integer> {
  @Override
  public boolean filter(Integer number) {
  return number >= 0;
  }
}

// [...]
DataSet<Integer> intNumbers = // [...]
DataSet<Integer> naturalNumbers = intNumbers.filter(new NaturalNumberFilter());

val intNumbers: DataSet[Int] = // [...] val naturalNumbers = intNumbers.filter { _ > 0 }

 naturalNumbers = intNumbers.filter(lambda x: x > 0)

重要信息： 系统假定该函数不会修改应用谓词的数据元。违反此假设可能会导致错误的结果。

元组数据集的 Projection

Project 转换删除或移动元组 DataSet 的 Tuple 字段。该 project(int...) 方法选择应由其索引保存的元组字段，并在输出元组中定义它们的顺序。

预测不需要定义用户函数。

以下代码显示了在 DataSet 上应用项目转换的不同方法：

Java
Scala
Python

DataSet<Tuple3<Integer, Double, String>> in = // [...]
// converts Tuple3<Integer, Double, String> into Tuple2<String, Integer>
DataSet<Tuple2<String, Integer>> out = in.project(2,0);

使用类型提示进行 Projection

请注意，Java 编译器无法推断 project 算子的返回类型。如果您对算子的结果调用另一个算子，则可能会导致问题， project 例如：

DataSet<Tuple5<String,String,String,String,String>> ds = ....
DataSet<Tuple1<String>> ds2 = ds.project(0).distinct(0);

通过提示返回类型的 project 算子可以克服此问题，如下所示：

DataSet<Tuple1<String>> ds2 = ds.<Tuple1<String>>project(0).distinct(0);

Not supported.

out = in.project(2,0);

分组数据集的转换

reduce 算子操作可以对分组数据集进行算子操作。指定用于分组的 Keys 可以通过多种方式完成：

关键表达
键选择器函数
一个或多个字段位置键（仅限元组数据集）
案例类字段（仅限案例类）

请查看 reduce 示例以了解如何指定分组键。

Reduce 分组数据集

应用于分组 DataSet 的 Reduce 转换使用用户定义的 reduce 函数将每个组 Reduce 为单个数据元。对于每组输入数据元，reduce 函数连续地将数据元对组合成一个数据元，直到每个组只剩下一个数据元。

请注意，对于 ReduceFunction 返回对象的被 Keys 化字段，应与输入值匹配。这是因为 reduce 是可隐式组合的，并且从组合算子发出的对象在传递给 reduce 算子时再次按键分组。

Reduce 由键表达式分组的 DataSet

键表达式指定 DataSet 的每个数据元的一个或多个字段。每个键表达式都是公共字段的名称或 getter 方法。点可用于向下钻取对象。关键表达式“*”选择所有字段。以下代码显示如何使用键表达式对 POJO DataSet 进行分组，并使用 reduce 函数对其进行缩减。

Java
Scala
Python

// some ordinary POJO
public class WC {
  public String word;
  public int count;
  // [...]
}

// ReduceFunction that sums Integer attributes of a POJO
public class WordCounter implements ReduceFunction<WC> {
  @Override
  public WC reduce(WC in1, WC in2) {
  return new WC(in1.word, in1.count + in2.count);
  }
}

// [...]
DataSet<WC> words = // [...]
DataSet<WC> wordCounts = words
             // DataSet grouping on field "word"
             .groupBy("word")
             // apply ReduceFunction on grouped DataSet
             .reduce(new WordCounter());

// some ordinary POJO class WC(val word: String, val count: Int) {
  def this() {
  this(null, -1)
  }
  // [...] }

val words: DataSet[WC] = // [...] val wordCounts = words.groupBy("word").reduce {
  (w1, w2) => new WC(w1.word, w1.count + w2.count)
}

Not supported.

Reduce 由 KeySelector 函数分组的 DataSet

键选择器函数从 DataSet 的每个数据元中提取键值。提取的键值用于对 DataSet 进行分组。以下代码显示如何使用键选择器函数对 POJO DataSet 进行分组，并使用 reduce 函数对其进行缩减。

Java
Scala
Python

// some ordinary POJO
public class WC {
  public String word;
  public int count;
  // [...]
}

// ReduceFunction that sums Integer attributes of a POJO
public class WordCounter implements ReduceFunction<WC> {
  @Override
  public WC reduce(WC in1, WC in2) {
  return new WC(in1.word, in1.count + in2.count);
  }
}

// [...]
DataSet<WC> words = // [...]
DataSet<WC> wordCounts = words
             // DataSet grouping on field "word"
             .groupBy(new SelectWord())
             // apply ReduceFunction on grouped DataSet
             .reduce(new WordCounter());

public class SelectWord implements KeySelector<WC, String> {
  @Override
  public String getKey(Word w) {
  return w.word;
  }
}

// some ordinary POJO class WC(val word: String, val count: Int) {
  def this() {
  this(null, -1)
  }
  // [...] }

val words: DataSet[WC] = // [...] val wordCounts = words.groupBy { _.word } reduce {
  (w1, w2) => new WC(w1.word, w1.count + w2.count)
}

class WordCounter(ReduceFunction):
  def reduce(self, in1, in2):
    return (in1[0], in1[1] + in2[1])

words = // [...]
wordCounts = words \
  .group_by(lambda x: x[0]) \
  .reduce(WordCounter())

Reduce 由字段位置键分组的 DataSet（仅限元组数据集）

字段位置键指定用作分组键的元组数据集的一个或多个字段。以下代码显示如何使用字段位置键并应用 reduce 函数

Java
Scala
Python

DataSet<Tuple3<String, Integer, Double>> tuples = // [...]
DataSet<Tuple3<String, Integer, Double>> reducedTuples = tuples
                     // group DataSet on first and second field of Tuple
                     .groupBy(0, 1)
                     // apply ReduceFunction on grouped DataSet
                     .reduce(new MyTupleReducer());

val tuples = DataSet[(String, Int, Double)] = // [...]
// group on the first and second Tuple field val reducedTuples = tuples.groupBy(0, 1).reduce { ... }

 reducedTuples = tuples.group_by(0, 1).reduce( ... )

按案例类字段分组的 DataSetReduce

使用 Case Classes 时，您还可以使用字段名称指定分组键：

Java
Scala
Python

Not supported.

case class MyClass(val a: String, b: Int, c: Double)
val tuples = DataSet[MyClass] = // [...]
// group on the first and second field val reducedTuples = tuples.groupBy("a", "b").reduce { ... }

Not supported.

GroupReduce 在分组数据集上

应用于分组 DataSet 的 GroupReduce 转换为每个组调用用户定义的 group-reduce 函数。这与 Reduce 之间的区别在于用户定义的函数会立即获得整个组。在组的所有数据元上使用 Iterable 调用该函数，并且可以返回任意数量的结果数据元。

由字段位置键分组的 DataSet 上的 GroupReduce（仅限元组数据集）

以下代码显示如何从按 Integer 分组的 DataSet 中删除重复的字符串。

Java
Scala
Python

public class DistinctReduce
     implements GroupReduceFunction<Tuple2<Integer, String>, Tuple2<Integer, String>> {

  @Override
  public void reduce(Iterable<Tuple2<Integer, String>> in, Collector<Tuple2<Integer, String>> out) {

  Set<String> uniqStrings = new HashSet<String>();
  Integer key = null;

  // add all strings of the group to the set
  for (Tuple2<Integer, String> t : in) {
    key = t.f0;
    uniqStrings.add(t.f1);
  }

  // emit all unique strings.
  for (String s : uniqStrings) {
    out.collect(new Tuple2<Integer, String>(key, s));
  }
  }
}

// [...]
DataSet<Tuple2<Integer, String>> input = // [...]
DataSet<Tuple2<Integer, String>> output = input
               .groupBy(0)      // group DataSet by the first tuple field
               .reduceGroup(new DistinctReduce());  // apply GroupReduceFunction

val input: DataSet[(Int, String)] = // [...] val output = input.groupBy(0).reduceGroup {
    (in, out: Collector[(Int, String)]) =>
    in.toSet foreach (out.collect)
  }

 class DistinctReduce(GroupReduceFunction):
   def reduce(self, iterator, collector):
   dic = dict()
   for value in iterator:
     dic[value[1]] = 1
   for key in dic.keys():
     collector.collect(key)

 output = data.group_by(0).reduce_group(DistinctReduce())

按键表达式，键选择器函数或案例类字段分组的 DataSet 上的 GroupReduce

类似于 Reduce 转换中的键表达式，键选择器函数和案例类字段的工作。

对已排序的组进行 GroupReduce

group-reduce 函数使用 Iterable 访问组的数据元。可选地，Iterable 可以按指定的顺序分发组的数据元。在许多情况下，这可以帮助降低用户定义的组 Reduce 函数的复杂性并提高其效率。

下面的代码显示了如何删除由 Integer 分组并按 String 排序的 DataSet 中的重复字符串的另一个示例。

Java
Scala
Python

// GroupReduceFunction that removes consecutive identical elements
public class DistinctReduce
     implements GroupReduceFunction<Tuple2<Integer, String>, Tuple2<Integer, String>> {

  @Override
  public void reduce(Iterable<Tuple2<Integer, String>> in, Collector<Tuple2<Integer, String>> out) {
  Integer key = null;
  String comp = null;

  for (Tuple2<Integer, String> t : in) {
    key = t.f0;
    String next = t.f1;

    // check if strings are different
    if (com == null || !next.equals(comp)) {
    out.collect(new Tuple2<Integer, String>(key, next));
    comp = next;
    }
  }
  }
}

// [...]
DataSet<Tuple2<Integer, String>> input = // [...]
DataSet<Double> output = input
             .groupBy(0)             // group DataSet by first field
             .sortGroup(1, Order.ASCENDING)    // sort groups on second tuple field
             .reduceGroup(new DistinctReduce());

val input: DataSet[(Int, String)] = // [...] val output = input.groupBy(0).sortGroup(1, Order.ASCENDING).reduceGroup {
    (in, out: Collector[(Int, String)]) =>
    var prev: (Int, String) = null
    for (t <- in) {
      if (prev == null || prev != t)
      out.collect(t)
      prev = t
    }
  }

 class DistinctReduce(GroupReduceFunction):
   def reduce(self, iterator, collector):
   dic = dict()
   for value in iterator:
     dic[value[1]] = 1
   for key in dic.keys():
     collector.collect(key)

 output = data.group_by(0).sort_group(1, Order.ASCENDING).reduce_group(DistinctReduce())

注意： 如果在 reduce 算子操作之前使用算子的基于排序的执行策略建立分组，则 GroupSort 通常是免费的。

可组合的 GroupReduce 函数

与 reduce 函数相比，group-reduce 函数不是可隐式组合的。为了使组合 - 缩减函数可组合，它必须实现 GroupCombineFunction 接口。

要点：接口的通用输入和输出类型 GroupCombineFunction 必须等于 GroupReduceFunction 以下示例中所示的通用输入类型：

Java
Scala
Python

// Combinable GroupReduceFunction that computes a sum.
public class MyCombinableGroupReducer implements
  GroupReduceFunction<Tuple2<String, Integer>, String>,
  GroupCombineFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>
{
  @Override
  public void reduce(Iterable<Tuple2<String, Integer>> in,
           Collector<String> out) {

  String key = null;
  int sum = 0;

  for (Tuple2<String, Integer> curr : in) {
    key = curr.f0;
    sum += curr.f1;
  }
  // concat key and sum and emit
  out.collect(key + "-" + sum);
  }

  @Override
  public void combine(Iterable<Tuple2<String, Integer>> in,
            Collector<Tuple2<String, Integer>> out) {
  String key = null;
  int sum = 0;

  for (Tuple2<String, Integer> curr : in) {
    key = curr.f0;
    sum += curr.f1;
  }
  // emit tuple with key and sum
  out.collect(new Tuple2<>(key, sum));
  }
}

// Combinable GroupReduceFunction that computes two sums. class MyCombinableGroupReducer
  extends GroupReduceFunction[(String, Int), String]
  with GroupCombineFunction[(String, Int), (String, Int)]
{
  override def reduce(
  in: java.lang.Iterable[(String, Int)],
  out: Collector[String]): Unit =
  {
  val r: (String, Int) =
    in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
  // concat key and sum and emit
  out.collect (r._1 + "-" + r._2)
  }

  override def combine(
  in: java.lang.Iterable[(String, Int)],
  out: Collector[(String, Int)]): Unit =
  {
  val r: (String, Int) =
    in.asScala.reduce( (a,b) => (a._1, a._2 + b._2) )
  // emit tuple with key and sum
  out.collect(r)
  }
}

 class GroupReduce(GroupReduceFunction):
   def reduce(self, iterator, collector):
   key, int_sum = iterator.next()
   for value in iterator:
     int_sum += value[1]
   collector.collect(key + "-" + int_sum))

   def combine(self, iterator, collector):
   key, int_sum = iterator.next()
   for value in iterator:
     int_sum += value[1]
   collector.collect((key, int_sum))

data.reduce_group(GroupReduce(), combinable=True)

GroupCombine 在分组数据集上

GroupCombine 变换是可组合 GroupReduceFunction 中的组合步骤的一般形式。从某种意义上说，它允许将输入类型组合 I 到任意输出类型 O 。相反，GroupReduce 中的组合步骤仅允许从输入类型 I 到输出类型的组合 I 。这是因为 GroupReduceFunction 中的 reduce 步骤需要输入类型 I 。

在一些应用中，期望在执行附加变换（例如，减小数据大小）之前将 DataSet 组合成中间格式。这可以通过 CombineGroup 转换以非常低的成本实现。

注意： 分组数据集上的 GroupCombine 在内存中使用贪婪策略执行，该策略可能不会一次处理所有数据，而是以多个步骤处理。它也可以在各个分区上执行，而无需像 GroupReduce 转换那样进行数据交换。这可能会导致部分结果。

以下示例演示了如何将 CombineGroup 转换用于 Slave WordCount 实现。

Java
Scala
Python

DataSet<String> input = [..] // The words received as input

DataSet<Tuple2<String, Integer>> combinedWords = input
  .groupBy(0) // group identical words
  .combineGroup(new GroupCombineFunction<String, Tuple2<String, Integer>() {

  public void combine(Iterable<String> words, Collector<Tuple2<String, Integer>>) { // combine
    String key = null;
    int count = 0;

    for (String word : words) {
      key = word;
      count++;
    }
    // emit tuple with word and count
    out.collect(new Tuple2(key, count));
  }
});

DataSet<Tuple2<String, Integer>> output = combinedWords
  .groupBy(0)                // group by words again
  .reduceGroup(new GroupReduceFunction() { // group reduce with full data exchange

  public void reduce(Iterable<Tuple2<String, Integer>>, Collector<Tuple2<String, Integer>>) {
    String key = null;
    int count = 0;

    for (Tuple2<String, Integer> word : words) {
      key = word;
      count++;
    }
    // emit tuple with word and count
    out.collect(new Tuple2(key, count));
  }
});

val input: DataSet[String] = [..] // The words received as input 
val combinedWords: DataSet[(String, Int)] = input
  .groupBy(0)
  .combineGroup {
  (words, out: Collector[(String, Int)]) =>
    var key: String = null
    var count = 0

    for (word <- words) {
      key = word
      count += 1
    }
    out.collect((key, count))
}

val output: DataSet[(String, Int)] = combinedWords
  .groupBy(0)
  .reduceGroup {
  (words, out: Collector[(String, Int)]) =>
    var key: String = null
    var sum = 0

    for ((word, sum) <- words) {
      key = word
      sum += count
    }
    out.collect((key, sum))
}

Not supported.

上面的替代 WordCount 实现演示了 GroupCombine 在执行 GroupReduce 转换之前如何组合单词。上面的例子只是一个概念证明。注意，组合步骤如何更改 DataSet 的类型，这通常需要在执行 GroupReduce 之前进行额外的 Map 转换。

聚合在分组元组数据集上

有一些常用的聚合算子操作经常使用。Aggregate 转换提供以下内置聚合函数：

Sum
Min，Sum
Max。

聚合转换只能应用于元组数据集，并且仅支持字段位置键进行分组。

以下代码显示如何对按字段位置键分组的 DataSet 应用聚合转换：

Java
Scala
Python

DataSet<Tuple3<Integer, String, Double>> input = // [...]
DataSet<Tuple3<Integer, String, Double>> output = input
                   .groupBy(1)    // group DataSet on second field
                   .aggregate(SUM, 0) // compute sum of the first field
                   .and(MIN, 2);    // compute minimum of the third field

val input: DataSet[(Int, String, Double)] = // [...] val output = input.groupBy(1).aggregate(SUM, 0).and(MIN, 2)

from flink.functions.Aggregation import Sum, Min

input = # [...]
output = input.group_by(1).aggregate(Sum, 0).and_agg(Min, 2)

要在 DataSet 上应用多个聚合，必须 .and() 在第一个聚合之后使用该函数，这意味着 .aggregate(SUM, 0).and(MIN, 2) 生成字段 0 的总和和原始 DataSet 的字段 2 的最小值。与此相反， .aggregate(SUM, 0).aggregate(MIN, 2) 将在聚合上应用聚合。在给定的示例中，在计算由字段 1 分组的字段 0 的总和之后，它将产生字段 2 的最小值。

注意： 将来会扩展聚合函数集。

MinBy / MaxBy 在 Grouped Tuple DataSet 上

MinBy（MaxBy）转换为每组元组选择一个元组。选定的元组是一个元组，其一个或多个指定字段的值最小（最大）。用于比较的字段必须是有效的关键字段，即可比较。如果多个元组具有最小（最大）字段值，则返回这些元组的任意元组。

下面的代码显示了如何选择具有最小值的元组，每个元组的字段 Integer 和 Double 字段具有相同的 String 值 DataSet<Tuple3<Integer, String, Double>> ：

Java
Scala
Python

DataSet<Tuple3<Integer, String, Double>> input = // [...]
DataSet<Tuple3<Integer, String, Double>> output = input
                   .groupBy(1)   // group DataSet on second field
                   .minBy(0, 2); // select tuple with minimum values for first and third field.

val input: DataSet[(Int, String, Double)] = // [...] val output: DataSet[(Int, String, Double)] = input
                   .groupBy(1)  // group DataSet on second field
                   .minBy(0, 2) // select tuple with minimum values for first and third field.

Not supported.

Reduce 完整的 DataSet

Reduce 转换将用户定义的 reduce 函数应用于 DataSet 的所有数据元。reduce 函数随后将数据元对组合成一个数据元，直到只剩下一个数据元。

以下代码显示了如何对 Integer DataSet 的所有数据元求和：

Java
Scala
Python

// ReduceFunction that sums Integers
public class IntSummer implements ReduceFunction<Integer> {
  @Override
  public Integer reduce(Integer num1, Integer num2) {
  return num1 + num2;
  }
}

// [...]
DataSet<Integer> intNumbers = // [...]
DataSet<Integer> sum = intNumbers.reduce(new IntSummer());

val intNumbers = env.fromElements(1,2,3)
val sum = intNumbers.reduce (_ + _)

 intNumbers = env.from_elements(1,2,3)
 sum = intNumbers.reduce(lambda x,y: x + y)

使用 Reduce 转换 Reduce 完整的 DataSet 意味着最终的 Reduce 算子操作不能并行完成。但是，reduce 函数可以自动组合，因此 Reduce 转换不会限制大多数用例的可伸缩性。

完整 DataSet 上的 GroupReduce

GroupReduce 转换在 DataSet 的所有数据元上应用用户定义的 group-reduce 函数。group-reduce 可以迭代 DataSet 的所有数据元并返回任意数量的结果数据元。

以下示例显示如何在完整 DataSet 上应用 GroupReduce 转换：

Java
Scala
Python

DataSet<Integer> input = // [...]
// apply a (preferably combinable) GroupReduceFunction to a DataSet
DataSet<Double> output = input.reduceGroup(new MyGroupReducer());

val input: DataSet[Int] = // [...] val output = input.reduceGroup(new MyGroupReducer())

 output = data.reduce_group(MyGroupReducer())

注意： 如果 group-reduce 函数不可组合，则无法并行完成对完整 DataSet 的 GroupReduce 转换。因此，这可能是计算密集型算子操作。请参阅上面的“可组合 GroupReduceFunctions”一节，了解如何实现可组合的 group-reduce 函数。

GroupCombine 在完整的 DataSet 上

完整 DataSet 上的 GroupCombine 与分组 DataSet 上的 GroupCombine 类似。数据在所有节点上分区，然后以贪婪的方式组合（即，只有一次合并到存储器中的数据）。

在完整的 Tuple DataSet 上聚合

有一些常用的聚合算子操作经常使用。Aggregate 转换提供以下内置聚合函数：

Sum
Min，Sum
Max。

聚合转换只能应用于元组数据集。

以下代码显示如何在完整 DataSet 上应用聚合转换：

Java
Scala
Python

DataSet<Tuple2<Integer, Double>> input = // [...]
DataSet<Tuple2<Integer, Double>> output = input
                   .aggregate(SUM, 0)  // compute sum of the first field
                   .and(MIN, 1);  // compute minimum of the second field

val input: DataSet[(Int, String, Double)] = // [...] val output = input.aggregate(SUM, 0).and(MIN, 2)

from flink.functions.Aggregation import Sum, Min

input = # [...]
output = input.aggregate(Sum, 0).and_agg(Min, 2)

注意： 扩展支持的聚合函数集在我们的路线图中。

完整的 Tuple DataSet 上的 MinBy / MaxBy

MinBy（MaxBy）转换从元组的 DataSet 中选择一个元组。选定的元组是一个元组，其一个或多个指定字段的值最小（最大）。用于比较的字段必须是有效的关键字段，即可比较。如果多个元组具有最小（最大）字段值，则返回这些元组的任意元组。

下面的代码演示如何选择与为最大值的元组 Integer ，并 Double 从一个领域 DataSet<Tuple3<Integer, String, Double>> ：

Java
Scala
Python

DataSet<Tuple3<Integer, String, Double>> input = // [...]
DataSet<Tuple3<Integer, String, Double>> output = input
                   .maxBy(0, 2); // select tuple with maximum values for first and third field.

val input: DataSet[(Int, String, Double)] = // [...] val output: DataSet[(Int, String, Double)] = input
                   .maxBy(0, 2) // select tuple with maximum values for first and third field.

Not supported.

Distinct

Distinct 转换计算源 DataSet 的不同数据元的 DataSet。以下代码从 DataSet 中删除所有重复的数据元：

Java
Scala
Python

DataSet<Tuple2<Integer, Double>> input = // [...]
DataSet<Tuple2<Integer, Double>> output = input.distinct();

val input: DataSet[(Int, String, Double)] = // [...] val output = input.distinct()

Not supported.

还可以使用以下方法更改 DataSet 中数据元的区别：

一个或多个字段位置键（仅限元组数据集），
键选择器函数，或
一个关键的表达。

DataSet<Tuple2<Integer, Double, String>> input = // [...]
DataSet<Tuple2<Integer, Double, String>> output = input.distinct(0,2);

val input: DataSet[(Int, Double, String)] = // [...] val output = input.distinct(0,2)

Not supported.

与 KeySelector 函数取 Distinct

Java
Scala
Python

private static class AbsSelector implements KeySelector<Integer, Integer> {
private static final long serialVersionUID = 1L;
  @Override
  public Integer getKey(Integer t) {
    return Math.abs(t);
  }
}
DataSet<Integer> input = // [...]
DataSet<Integer> output = input.distinct(new AbsSelector());

val input: DataSet[Int] = // [...] val output = input.distinct {x => Math.abs(x)}

Not supported.

用 Key 表达式取 Distinct

Java
Scala
Python

// some ordinary POJO
public class CustomType {
  public String aName;
  public int aNumber;
  // [...]
}

DataSet<CustomType> input = // [...]
DataSet<CustomType> output = input.distinct("aName", "aNumber");

// some ordinary POJO case class CustomType(aName : String, aNumber : Int) { }

val input: DataSet[CustomType] = // [...] val output = input.distinct("aName", "aNumber")

Not supported.

也可以通过通配符指示使用所有字段：

Java
Scala
Python

DataSet<CustomType> input = // [...]
DataSet<CustomType> output = input.distinct("*");

// some ordinary POJO val input: DataSet[CustomType] = // [...] val output = input.distinct("_")

Not supported.

Join

Join 转换将两个 DataSet 连接到一个 DataSet 中。两个 DataSet 的数据元连接在一个或多个可以使用的键上

一个关键的表达
键选择器函数
一个或多个字段位置键（仅限元组数据集）。
案例类字段

有几种不同的方法可以执行 Join 转换，如下所示。

默认 Join（关联 Tuple2）

默认的 Join 转换生成一个包含两个字段的新 Tuple DataSet。每个元组保存第一个元组字段中第一个输入 DataSet 的连接数据元和第二个字段中第二个输入 DataSet 的匹配数据元。

以下代码显示使用字段位置键的默认 Join 转换：

Java
Scala
Python

public static class User { public String name; public int zip; }
public static class Store { public Manager mgr; public int zip; }
DataSet<User> input1 = // [...]
DataSet<Store> input2 = // [...]
// result dataset is typed as Tuple2
DataSet<Tuple2<User, Store>>
      result = input1.join(input2)
               .where("zip")     // key of the first input (users)
               .equalTo("zip");  // key of the second input (stores)

val input1: DataSet[(Int, String)] = // [...] val input2: DataSet[(Double, Int)] = // [...] val result = input1.join(input2).where(0).equalTo(1)

 result = input1.join(input2).where(0).equal_to(1)

关联 Join 函数

Join 转换还可以调用用户定义的连接函数来处理连接元组。连接函数接收第一个输入 DataSet 的一个数据元和第二个输入 DataSet 的一个数据元，并返回一个数据元。

以下代码使用键选择器函数执行 DataSet 与自定义 java 对象和 Tuple DataSet 的连接，并显示如何使用用户定义的连接函数：

Java
Scala
Python

// some POJO
public class Rating {
  public String name;
  public String category;
  public int points;
}

// Join function that joins a custom POJO with a Tuple
public class PointWeighter
     implements JoinFunction<Rating, Tuple2<String, Double>, Tuple2<String, Double>> {

  @Override
  public Tuple2<String, Double> join(Rating rating, Tuple2<String, Double> weight) {
  // multiply the points and rating and construct a new output tuple
  return new Tuple2<String, Double>(rating.name, rating.points * weight.f1);
  }
}

DataSet<Rating> ratings = // [...]
DataSet<Tuple2<String, Double>> weights = // [...]
DataSet<Tuple2<String, Double>>
      weightedRatings =
      ratings.join(weights)

           // key of the first input
           .where("category")

           // key of the second input
           .equalTo("f0")

           // applying the JoinFunction on joining pairs
           .with(new PointWeighter());

case class Rating(name: String, category: String, points: Int)

val ratings: DataSet[Ratings] = // [...] val weights: DataSet[(String, Double)] = // [...] 
val weightedRatings = ratings.join(weights).where("category").equalTo(0) {
  (rating, weight) => (rating.name, rating.points * weight._2)
}

 class PointWeighter(JoinFunction):
   def join(self, rating, weight):
   return (rating[0], rating[1] * weight[1])
     if value1[3]:

 weightedRatings =
   ratings.join(weights).where(0).equal_to(0). \
   with(new PointWeighter());

关联 Flat-Join 函数

类似于 Map 和 FlatMap，FlatJoin 的行为与 Join 相同，但它不是返回一个数据元，而是返回（收集），零，一个或多个数据元。

Java
Scala
Python

public class PointWeighter
     implements FlatJoinFunction<Rating, Tuple2<String, Double>, Tuple2<String, Double>> {
  @Override
  public void join(Rating rating, Tuple2<String, Double> weight,
    Collector<Tuple2<String, Double>> out) {
  if (weight.f1 > 0.1) {
    out.collect(new Tuple2<String, Double>(rating.name, rating.points * weight.f1));
  }
  }
}

DataSet<Tuple2<String, Double>>
      weightedRatings =
      ratings.join(weights) // [...]

case class Rating(name: String, category: String, points: Int)

val ratings: DataSet[Ratings] = // [...] val weights: DataSet[(String, Double)] = // [...] 
val weightedRatings = ratings.join(weights).where("category").equalTo(0) {
  (rating, weight, out: Collector[(String, Double)]) =>
  if (weight._2 > 0.1) out.collect(rating.name, rating.points * weight._2)
}

Not supported.

关联 Projection（仅限 Java / Python）

Join 变换可以使用 Projection 构造结果元组，如下所示：

Java
Scala
Python

DataSet<Tuple3<Integer, Byte, String>> input1 = // [...]
DataSet<Tuple2<Integer, Double>> input2 = // [...]
DataSet<Tuple4<Integer, String, Double, Byte>>
      result =
      input1.join(input2)
          // key definition on first DataSet using a field position key
          .where(0)
          // key definition of second DataSet using a field position key
          .equalTo(0)
          // select and reorder fields of matching tuples
          .projectFirst(0,2).projectSecond(1).projectFirst(1);

projectFirst(int...) 并 projectSecond(int...) 选择应组合成输出元组的第一个和第二个连接输入的字段。索引的顺序定义输出元组中的字段顺序。连接 Projection 也适用于非元组数据集。在这种情况下， projectFirst() 或者 projectSecond() 必须不带参数调用才能将连接数据元添加到输出元组。

Not supported.

 result = input1.join(input2).where(0).equal_to(0) \
  .project_first(0,2).project_second(1).project_first(1);

project_first(int...) and project_second(int...) select the fields of the first and second joined input that should be assembled into an output Tuple. The order of indexes defines the order of fields in the output tuple. The join projection works also for non-Tuple DataSets. In this case, project_first() or project_second() must be called without arguments to add a joined element to the output Tuple.

关联 DataSet Size 提示

为了引导优化器选择正确的执行策略，您可以提示要关联的 DataSet 的大小，如下所示：

Java
Scala
Python

DataSet<Tuple2<Integer, String>> input1 = // [...]
DataSet<Tuple2<Integer, String>> input2 = // [...]

DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<Integer, String>>>
      result1 =
      // hint that the second DataSet is very small
      input1.joinWithTiny(input2)
          .where(0)
          .equalTo(0);

DataSet<Tuple2<Tuple2<Integer, String>, Tuple2<Integer, String>>>
      result2 =
      // hint that the second DataSet is very large
      input1.joinWithHuge(input2)
          .where(0)
          .equalTo(0);

val input1: DataSet[(Int, String)] = // [...] val input2: DataSet[(Int, String)] = // [...] 
// hint that the second DataSet is very small val result1 = input1.joinWithTiny(input2).where(0).equalTo(0)

// hint that the second DataSet is very large val result1 = input1.joinWithHuge(input2).where(0).equalTo(0)

 #hint that the second DataSet is very small
 result1 = input1.join_with_tiny(input2).where(0).equal_to(0)

 #hint that the second DataSet is very large
 result1 = input1.join_with_huge(input2).where(0).equal_to(0)

关联算法提示

Flink 运行时可以以各种方式执行连接。在不同情况下，每种可能的方式都优于其他方式。系统会尝试自动选择合理的方式，但允许您手动选择策略，以防您想要强制执行连接的特定方式。

Java
Scala
Python

DataSet<SomeType> input1 = // [...]
DataSet<AnotherType> input2 = // [...]

DataSet<Tuple2<SomeType, AnotherType> result =
    input1.join(input2, JoinHint.BROADCAST_HASH_FIRST)
      .where("id").equalTo("key");

val input1: DataSet[SomeType] = // [...] val input2: DataSet[AnotherType] = // [...] 
// hint that the second DataSet is very small val result1 = input1.join(input2, JoinHint.BROADCAST_HASH_FIRST).where("id").equalTo("key")

Not supported.

以下提示可用：

OPTIMIZER_CHOOSES ：相当于不提供任何提示，将选择留给系统。
BROADCAST_HASH_FIRST ：广播第一个输入并从中构建哈希表，由第二个输入探测。如果第一个输入非常小，这是一个很好的策略。
BROADCAST_HASH_SECOND ：广播第二个输入并从中构建哈希表，由第一个输入探测。如果第二个输入非常小，这是一个很好的策略。
REPARTITION_HASH_FIRST ：系统分区（shuffle）每个输入（除非输入已经分区）并从第一个输入构建哈希表。如果第一个输入小于第二个输入，则此策略很好，但两个输入仍然很大。注意：这是系统使用的默认回退策略，如果不能进行大小估计，并且不能重新使用预先存在的分区和排序顺序。
REPARTITION_HASH_SECOND ：系统分区（shuffle）每个输入（除非输入已经分区）并从第二个输入构建哈希表。如果第二个输入小于第一个输入，则此策略很好，但两个输入仍然很大。
REPARTITION_SORT_MERGE ：系统对每个输入进行分区（shuffle）（除非输入已经分区）并对每个输入进行排序（除非它已经排序）。输入通过已排序输入的流合并来连接。如果已经对一个或两个输入进行了排序，则此策略很好。

Outer Join

OuterJoin 转换在两个数据集上执行左，右或全外连接。外连接类似于常规（内部）连接，并创建在其键上相等的所有数据元对。此外，如果在另一侧没有找到匹配的 Keys，则保存“外部”侧（左侧，右侧或两者都满）的记录。匹配数据元对（或一个数据元和 null 另一个输入的值）被赋予 a JoinFunction 以将该对数据元转换为单个数据元，或者 FlatJoinFunction 将该数据元对转换为任意多个（包括无）数据元。

两个 DataSet 的数据元连接在一个或多个可以使用的键上

一个关键的表达
键选择器函数
一个或多个字段位置键（仅限元组数据集）。
案例类字段

OuterJoins 仅支持 Java 和 Scala DataSet API。

具有连接函数的 OuterJoin

OuterJoin 转换调用用户定义的连接函数来处理连接元组。连接函数接收第一个输入 DataSet 的一个数据元和第二个输入 DataSet 的一个数据元，并返回一个数据元。根据外连接的类型（left，right，full），连接函数的两个输入数据元之一可以是 null 。

以下代码使用键选择器函数执行 DataSet 与自定义 java 对象和 Tuple DataSet 的左外连接，并显示如何使用用户定义的连接函数：

Java
Scala
Python

// some POJO
public class Rating {
  public String name;
  public String category;
  public int points;
}

// Join function that joins a custom POJO with a Tuple
public class PointAssigner
     implements JoinFunction<Tuple2<String, String>, Rating, Tuple2<String, Integer>> {

  @Override
  public Tuple2<String, Integer> join(Tuple2<String, String> movie, Rating rating) {
  // Assigns the rating points to the movie.
  // NOTE: rating might be null
  return new Tuple2<String, Double>(movie.f0, rating == null ? -1 : rating.points;
  }
}

DataSet<Tuple2<String, String>> movies = // [...]
DataSet<Rating> ratings = // [...]
DataSet<Tuple2<String, Integer>>
      moviesWithPoints =
      movies.leftOuterJoin(ratings)

           // key of the first input
           .where("f0")

           // key of the second input
           .equalTo("name")

           // applying the JoinFunction on joining pairs
           .with(new PointAssigner());

case class Rating(name: String, category: String, points: Int)

val movies: DataSet[(String, String)] = // [...] val ratings: DataSet[Ratings] = // [...] 
val moviesWithPoints = movies.leftOuterJoin(ratings).where(0).equalTo("name") {
  (movie, rating) => (movie._1, if (rating == null) -1 else rating.points)
}

Not supported.

具有 Flat-Join 函数的外部连接

类似于 Map 和 FlatMap，具有 Flat-Join 函数的 OuterJoin 与具有连接函数的 OuterJoin 的行为方式相同，但它不返回一个数据元，而是返回（收集），零个，一个或多个数据元。

Java
Scala
Python

public class PointAssigner
     implements FlatJoinFunction<Tuple2<String, String>, Rating, Tuple2<String, Integer>> {
  @Override
  public void join(Tuple2<String, String> movie, Rating rating
  Collector<Tuple2<String, Integer>> out) {
  if (rating == null ) {
  out.collect(new Tuple2<String, Integer>(movie.f0, -1));
  } else if (rating.points < 10) {
  out.collect(new Tuple2<String, Integer>(movie.f0, rating.points));
  } else {
  // do not emit
  }
}

DataSet<Tuple2<String, Integer>>
      moviesWithPoints =
      movies.leftOuterJoin(ratings) // [...]

Not supported.

Not supported.

关联算法提示

Flink 运行时可以以各种方式执行外连接。在不同情况下，每种可能的方式都优于其他方式。系统会尝试自动选择合理的方式，但允许您手动选择策略，以防您想要强制执行外连接的特定方式。

Java
Scala
Python

DataSet<SomeType> input1 = // [...]
DataSet<AnotherType> input2 = // [...]

DataSet<Tuple2<SomeType, AnotherType> result1 =
    input1.leftOuterJoin(input2, JoinHint.REPARTITION_SORT_MERGE)
      .where("id").equalTo("key");

DataSet<Tuple2<SomeType, AnotherType> result2 =
    input1.rightOuterJoin(input2, JoinHint.BROADCAST_HASH_FIRST)
      .where("id").equalTo("key");

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")

Not supported.

以下提示可用。

OPTIMIZER_CHOOSES ：相当于不提供任何提示，将选择留给系统。
BROADCAST_HASH_FIRST ：广播第一个输入并从中构建哈希表，由第二个输入探测。如果第一个输入非常小，这是一个很好的策略。
BROADCAST_HASH_SECOND ：广播第二个输入并从中构建哈希表，由第一个输入探测。如果第二个输入非常小，这是一个很好的策略。
REPARTITION_HASH_FIRST ：系统分区（shuffle）每个输入（除非输入已经分区）并从第一个输入构建哈希表。如果第一个输入小于第二个输入，则此策略很好，但两个输入仍然很大。
REPARTITION_HASH_SECOND ：系统分区（shuffle）每个输入（除非输入已经分区）并从第二个输入构建哈希表。如果第二个输入小于第一个输入，则此策略很好，但两个输入仍然很大。
REPARTITION_SORT_MERGE ：系统对每个输入进行分区（shuffle）（除非输入已经分区）并对每个输入进行排序（除非它已经排序）。输入通过已排序输入的流合并来连接。如果已经对一个或两个输入进行了排序，则此策略很好。

注意： 并非所有外部联接类型都支持所有执行策略。

LeftOuterJoin 支持：
- OPTIMIZER_CHOOSES
- BROADCAST_HASH_SECOND
- REPARTITION_HASH_SECOND
- REPARTITION_SORT_MERGE
RightOuterJoin 支持：
- OPTIMIZER_CHOOSES
- BROADCAST_HASH_FIRST
- REPARTITION_HASH_FIRST
- REPARTITION_SORT_MERGE
FullOuterJoin 支持：
- OPTIMIZER_CHOOSES
- REPARTITION_SORT_MERGE

交叉

交叉转换将两个 DataSet 组合到一个 DataSet 中。它构建了两个输入 DataSet 的数据元的所有成对组合，即它构建了一个笛卡尔积。交叉变换要么在每对数据元上调用用户定义的交叉函数，要么输出 Tuple2。两种模式如下所示。

注：十字是一个潜在的非常计算密集型算子操作它甚至可以挑战大的计算集群！

与用户定义的函数交叉

交叉转换可以调用用户定义的交叉函数。交叉函数接收第一个输入的一个数据元和第二个输入的一个数据元，并返回一个结果数据元。

以下代码显示如何使用交叉函数在两个 DataSet 上应用 Cross 转换：

Java
Scala
Python

public class Coord {
  public int id;
  public int x;
  public int y;
}

// CrossFunction computes the Euclidean distance between two Coord objects.
public class EuclideanDistComputer
     implements CrossFunction<Coord, Coord, Tuple3<Integer, Integer, Double>> {

  @Override
  public Tuple3<Integer, Integer, Double> cross(Coord c1, Coord c2) {
  // compute Euclidean distance of coordinates
  double dist = sqrt(pow(c1.x - c2.x, 2) + pow(c1.y - c2.y, 2));
  return new Tuple3<Integer, Integer, Double>(c1.id, c2.id, dist);
  }
}

DataSet<Coord> coords1 = // [...]
DataSet<Coord> coords2 = // [...]
DataSet<Tuple3<Integer, Integer, Double>>
      distances =
      coords1.cross(coords2)
           // apply CrossFunction
           .with(new EuclideanDistComputer());

与 Projection 交叉

交叉变换还可以使用 Projection 构造结果元组，如下所示：

DataSet<Tuple3<Integer, Byte, String>> input1 = // [...]
DataSet<Tuple2<Integer, Double>> input2 = // [...]
DataSet<Tuple4<Integer, Byte, Integer, Double>
      result =
      input1.cross(input2)
          // select and reorder fields of matching tuples
          .projectSecond(0).projectFirst(1,0).projectSecond(1);

交叉 Projection 中的字段选择与连接结果的 Projection 中的工作方式相同。

case class Coord(id: Int, x: Int, y: Int)

val coords1: DataSet[Coord] = // [...] val coords2: DataSet[Coord] = // [...] 
val distances = coords1.cross(coords2) {
  (c1, c2) =>
  val dist = sqrt(pow(c1.x - c2.x, 2) + pow(c1.y - c2.y, 2))
  (c1.id, c2.id, dist)
}

 class Euclid(CrossFunction):
   def cross(self, c1, c2):
   return (c1[0], c2[0], sqrt(pow(c1[1] - c2.[1], 2) + pow(c1[2] - c2[2], 2)))

 distances = coords1.cross(coords2).using(Euclid())

Cross with Projection

A Cross transformation can also construct result tuples using a projection as shown here:

result = input1.cross(input2).projectFirst(1,0).projectSecond(0,1);

The field selection in a Cross projection works the same way as in the projection of Join results.

与 DataSet 大小提示交叉

为了引导优化器选择正确的执行策略，您可以提示要交叉的 DataSet 的大小，如下所示：

Java
Scala
Python

DataSet<Tuple2<Integer, String>> input1 = // [...]
DataSet<Tuple2<Integer, String>> input2 = // [...]

DataSet<Tuple4<Integer, String, Integer, String>>
      udfResult =
          // hint that the second DataSet is very small
      input1.crossWithTiny(input2)
          // apply any Cross function (or projection)
          .with(new MyCrosser());

DataSet<Tuple3<Integer, Integer, String>>
      projectResult =
          // hint that the second DataSet is very large
      input1.crossWithHuge(input2)
          // apply a projection (or any Cross function)
          .projectFirst(0,1).projectSecond(1);

val input1: DataSet[(Int, String)] = // [...] val input2: DataSet[(Int, String)] = // [...] 
// hint that the second DataSet is very small val result1 = input1.crossWithTiny(input2)

// hint that the second DataSet is very large val result1 = input1.crossWithHuge(input2)

 #hint that the second DataSet is very small
 result1 = input1.cross_with_tiny(input2)

 #hint that the second DataSet is very large
 result1 = input1.cross_with_huge(input2)

CoGroup

CoGroup 转换共同处理两个 DataSet 的组。两个 DataSet 都在定义的 Keys 上分组，并且共享相同 Keys 的两个 DataSet 的组被一起交给用户定义的共同组函数。如果对于特定键，只有一个 DataSet 具有组，则使用该组和空组调用 co-group 函数。共同组函数可以单独迭代两个组的数据元并返回任意数量的结果数据元。

与 Reduce，GroupReduce 和 Join 类似，可以使用不同的键选择方法定义键。

DataSet 上的 CoGroup

Java
Scala
Python

该示例显示如何按字段位置键进行分组（仅限元组数据集）。您可以使用 Pojo 类型和键表达式执行相同的算子操作。

// Some CoGroupFunction definition
class MyCoGrouper
     implements CoGroupFunction<Tuple2<String, Integer>, Tuple2<String, Double>, Double> {

  @Override
  public void coGroup(Iterable<Tuple2<String, Integer>> iVals,
            Iterable<Tuple2<String, Double>> dVals,
            Collector<Double> out) {

  Set<Integer> ints = new HashSet<Integer>();

  // add all Integer values in group to set
  for (Tuple2<String, Integer>> val : iVals) {
    ints.add(val.f1);
  }

  // multiply each Double value with each unique Integer values of group
  for (Tuple2<String, Double> val : dVals) {
    for (Integer i : ints) {
    out.collect(val.f1 * i);
    }
  }
  }
}

// [...]
DataSet<Tuple2<String, Integer>> iVals = // [...]
DataSet<Tuple2<String, Double>> dVals = // [...]
DataSet<Double> output = iVals.coGroup(dVals)
             // group first DataSet on first tuple field
             .where(0)
             // group second DataSet on first tuple field
             .equalTo(0)
             // apply CoGroup function on each pair of groups
             .with(new MyCoGrouper());

val iVals: DataSet[(String, Int)] = // [...] val dVals: DataSet[(String, Double)] = // [...] 
val output = iVals.coGroup(dVals).where(0).equalTo(0) {
  (iVals, dVals, out: Collector[Double]) =>
  val ints = iVals map { _._2 } toSet

  for (dVal <- dVals) {
    for (i <- ints) {
    out.collect(dVal._2 * i)
    }
  }
}

 class CoGroup(CoGroupFunction):
   def co_group(self, ivals, dvals, collector):
   ints = dict()
   # add all Integer values in group to set
   for value in ivals:
     ints[value[1]] = 1
   # multiply each Double value with each unique Integer values of group
   for value in dvals:
     for i in ints.keys():
     collector.collect(value[1] * i)

 output = ivals.co_group(dvals).where(0).equal_to(0).using(CoGroup())

Union

生成两个 DataSet 的并集，它们必须属于同一类型。可以使用多个联合调用实现两个以上 DataSet 的并集，如下所示：

Java
Scala
Python

DataSet<Tuple2<String, Integer>> vals1 = // [...]
DataSet<Tuple2<String, Integer>> vals2 = // [...]
DataSet<Tuple2<String, Integer>> vals3 = // [...]
DataSet<Tuple2<String, Integer>> unioned = vals1.union(vals2).union(vals3);

val vals1: DataSet[(String, Int)] = // [...] val vals2: DataSet[(String, Int)] = // [...] val vals3: DataSet[(String, Int)] = // [...] 
val unioned = vals1.union(vals2).union(vals3)

 unioned = vals1.union(vals2).union(vals3)

Rebalance

均匀地 Rebalance DataSet 的并行分区以消除数据偏斜。

Java
Scala
Python

DataSet<String> in = // [...]
// rebalance DataSet and apply a Map transformation.
DataSet<Tuple2<String, String>> out = in.rebalance()
                    .map(new Mapper());

val in: DataSet[String] = // [...]
// rebalance DataSet and apply a Map transformation. val out = in.rebalance().map { ... }

Not supported.

Hash-Partition

散列分区给定键上的 DataSet。键可以指定为位置键，表达式键和键选择器函数（有关如何指定键，请参阅 Reduce 示例）。

Java
Scala
Python

DataSet<Tuple2<String, Integer>> in = // [...]
// hash-partition DataSet by String value and apply a MapPartition transformation.
DataSet<Tuple2<String, String>> out = in.partitionByHash(0)
                    .mapPartition(new PartitionMapper());

val in: DataSet[(String, Int)] = // [...]
// hash-partition DataSet by String value and apply a MapPartition transformation. val out = in.partitionByHash(0).mapPartition { ... }

Not supported.

Range-Partition

对给定键的 DataSet 进行 Range-Partition。键可以指定为位置键，表达式键和键选择器函数（有关如何指定键，请参阅 Reduce 示例）。

Java
Scala
Python

DataSet<Tuple2<String, Integer>> in = // [...]
// range-partition DataSet by String value and apply a MapPartition transformation.
DataSet<Tuple2<String, String>> out = in.partitionByRange(0)
                    .mapPartition(new PartitionMapper());

val in: DataSet[(String, Int)] = // [...]
// range-partition DataSet by String value and apply a MapPartition transformation. val out = in.partitionByRange(0).mapPartition { ... }

Not supported.

Sort Partition

本地按指定顺序对指定字段上的 DataSet 的所有分区进行排序。可以将字段指定为字段表达式或字段位置（有关如何指定键，请参阅 Reduce 示例）。可以通过链接 sortPartition() 调用在多个字段上对分区进行排序。

Java
Scala
Python

DataSet<Tuple2<String, Integer>> in = // [...]
// Locally sort partitions in ascending order on the second String field and
// in descending order on the first String field.
// Apply a MapPartition transformation on the sorted partitions.
DataSet<Tuple2<String, String>> out = in.sortPartition(1, Order.ASCENDING)
                    .sortPartition(0, Order.DESCENDING)
                    .mapPartition(new PartitionMapper());

val in: DataSet[(String, Int)] = // [...]
// Locally sort partitions in ascending order on the second String field and
// in descending order on the first String field.
// Apply a MapPartition transformation on the sorted partitions. val out = in.sortPartition(1, Order.ASCENDING)
      .sortPartition(0, Order.DESCENDING)
      .mapPartition { ... }

Not supported.

First-n

返回 DataSet 的前 n 个（任意）数据元。First-n 可以应用于常规 DataSet，分组 DataSet 或分组排序 DataSet。可以将分组键指定为键选择器函数或字段位置键（有关如何指定键，请参阅 Reduce 示例）。

Java
Scala
Python

DataSet<Tuple2<String, Integer>> in = // [...]
// Return the first five (arbitrary) elements of the DataSet
DataSet<Tuple2<String, Integer>> out1 = in.first(5);

// Return the first two (arbitrary) elements of each String group
DataSet<Tuple2<String, Integer>> out2 = in.groupBy(0)
                      .first(2);

// Return the first three elements of each String group ordered by the Integer field
DataSet<Tuple2<String, Integer>> out3 = in.groupBy(0)
                      .sortGroup(1, Order.ASCENDING)
                      .first(3);

val in: DataSet[(String, Int)] = // [...]
// Return the first five (arbitrary) elements of the DataSet val out1 = in.first(5)

// Return the first two (arbitrary) elements of each String group val out2 = in.groupBy(0).first(2)

// Return the first three elements of each String group ordered by the Integer field val out3 = in.groupBy(0).sortGroup(1, Order.ASCENDING).first(3)

Not supported.

分享到QQ

分享到微博

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

列表为空，暂无数据