47.md

# 数据集转换

> 译者：[flink.sojb.cn](https://flink.sojb.cn/)


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

有关在具有密集索引的数据集中压缩数据元，请参阅[Zip数据元指南](zip_elements_guide.html)。

### Map

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

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

*   [**Java**](#tab_java_0)
*   [**Scala**](#tab_scala_0)
*   [**Python**](#tab_python_0)



```
// 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**](#tab_java_1)
*   [**Scala**](#tab_scala_1)
*   [**Python**](#tab_python_1)



```
// 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**](#tab_java_2)
*   [**Scala**](#tab_scala_2)
*   [**Python**](#tab_python_2)



```
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**](#tab_java_3)
*   [**Scala**](#tab_scala_3)
*   [**Python**](#tab_python_3)



```
// 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**](#tab_java_4)
*   [**Scala**](#tab_scala_4)
*   [**Python**](#tab_python_4)



```
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**](#tab_java_5)
*   [**Scala**](#tab_scala_5)
*   [**Python**](#tab_python_5)



```
// 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**](#tab_java_6)
*   [**Scala**](#tab_scala_6)
*   [**Python**](#tab_python_6)



```
// 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**](#tab_java_7)
*   [**Scala**](#tab_scala_7)
*   [**Python**](#tab_python_7)



```
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**](#tab_java_8)
*   [**Scala**](#tab_scala_8)
*   [**Python**](#tab_python_8)



```
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**](#tab_java_9)
*   [**Scala**](#tab_scala_9)
*   [**Python**](#tab_python_9)



```
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_转换中的[键表达式](#reduce-on-dataset-grouped-by-key-expression)， [键选择器函数](#reduce-on-dataset-grouped-by-keyselector-function)和[案例类字段的](#reduce-on-dataset-grouped-by-case-class-fields)工作。

#### 对已排序的组进行GroupReduce

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

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

*   [**Java**](#tab_java_10)
*   [**Scala**](#tab_scala_10)
*   [**Python**](#tab_python_10)



```
// 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**](#tab_java_11)
*   [**Scala**](#tab_scala_11)
*   [**Python**](#tab_python_11)



```
// 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**](#tab_java_12)
*   [**Scala**](#tab_scala_12)
*   [**Python**](#tab_python_12)



```
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**](#tab_java_13)
*   [**Scala**](#tab_scala_13)
*   [**Python**](#tab_python_13)



```
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&lt;Tuple3&lt;Integer, String, Double&gt;&gt;`：

*   [**Java**](#tab_java_14)
*   [**Scala**](#tab_scala_14)
*   [**Python**](#tab_python_14)



```
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**](#tab_java_15)
*   [**Scala**](#tab_scala_15)
*   [**Python**](#tab_python_15)



```
// 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**](#tab_java_16)
*   [**Scala**](#tab_scala_16)
*   [**Python**](#tab_python_16)



```
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**](#tab_java_17)
*   [**Scala**](#tab_scala_17)
*   [**Python**](#tab_python_17)



```
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&lt;Tuple3&lt;Integer, String, Double&gt;&gt;`：

*   [**Java**](#tab_java_18)
*   [**Scala**](#tab_scala_18)
*   [**Python**](#tab_python_18)



```
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**](#tab_java_19)
*   [**Scala**](#tab_scala_19)
*   [**Python**](#tab_python_19)



```
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中数据元的区别：

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

#### 与列位置Keys取Distinct

*   [**Java**](#tab_java_20)
*   [**Scala**](#tab_scala_20)
*   [**Python**](#tab_python_20)



```
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**](#tab_java_21)
*   [**Scala**](#tab_scala_21)
*   [**Python**](#tab_python_21)



```
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**](#tab_java_22)
*   [**Scala**](#tab_scala_22)
*   [**Python**](#tab_python_22)



```
// 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**](#tab_java_23)
*   [**Scala**](#tab_scala_23)
*   [**Python**](#tab_python_23)



```
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**](#tab_java_24)
*   [**Scala**](#tab_scala_24)
*   [**Python**](#tab_python_24)



```
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**](#tab_java_25)
*   [**Scala**](#tab_scala_25)
*   [**Python**](#tab_python_25)



```
// 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**](#tab_java_26)
*   [**Scala**](#tab_scala_26)
*   [**Python**](#tab_python_26)



```
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**](#tab_java_27)
*   [**Scala**](#tab_scala_27)
*   [**Python**](#tab_python_27)



```
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**](#tab_java_28)
*   [**Scala**](#tab_scala_28)
*   [**Python**](#tab_python_28)



```
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**](#tab_java_29)
*   [**Scala**](#tab_scala_29)
*   [**Python**](#tab_python_29)



```
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**](#tab_java_30)
*   [**Scala**](#tab_scala_30)
*   [**Python**](#tab_python_30)



```
// 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**](#tab_java_31)
*   [**Scala**](#tab_scala_31)
*   [**Python**](#tab_python_31)



```
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**](#tab_java_32)
*   [**Scala**](#tab_scala_32)
*   [**Python**](#tab_python_32)



```
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**](#tab_java_33)
*   [**Scala**](#tab_scala_33)
*   [**Python**](#tab_python_33)



```
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**](#tab_java_34)
*   [**Scala**](#tab_scala_34)
*   [**Python**](#tab_python_34)



```
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**](#tab_java_35)
*   [**Scala**](#tab_scala_35)
*   [**Python**](#tab_python_35)

该示例显示如何按字段位置键进行分组（仅限元组数据集）。您可以使用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**](#tab_java_36)
*   [**Scala**](#tab_scala_36)
*   [**Python**](#tab_python_36)



```
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**](#tab_java_37)
*   [**Scala**](#tab_scala_37)
*   [**Python**](#tab_python_37)



```
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示例](#reduce-on-grouped-dataset)）。

*   [**Java**](#tab_java_38)
*   [**Scala**](#tab_scala_38)
*   [**Python**](#tab_python_38)



```
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示例](#reduce-on-grouped-dataset)）。

*   [**Java**](#tab_java_39)
*   [**Scala**](#tab_scala_39)
*   [**Python**](#tab_python_39)



```
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示例](#reduce-on-grouped-dataset)）。可以通过链接`sortPartition()`调用在多个字段上对分区进行排序。

*   [**Java**](#tab_java_40)
*   [**Scala**](#tab_scala_40)
*   [**Python**](#tab_python_40)



```
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示例](#reduce-on-grouped-dataset)）。

*   [**Java**](#tab_java_41)
*   [**Scala**](#tab_scala_41)
*   [**Python**](#tab_python_41)



```
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.
```