> out = in.project(2,0).types(String.class, Integer.class);
-```
+~~~
-### Transformations on grouped DataSet
+### Transformations on Grouped DataSet
The reduce operations can operate on grouped data sets. Specifying the key to
be used for grouping can be done in two ways:
-- a `KeySelector` function or
-- one or more field position keys (`Tuple` `DataSet` only).
+- a key-selector function or
+- one or more field position keys (Tuple DataSet only).
+- Case Class fields (Case Classes only).
Please look at the reduce examples to see how the grouping keys are specified.
-### Reduce on grouped DataSet
+### Reduce on Grouped DataSet
-A Reduce transformation that is applied on a grouped `DataSet` reduces each group to a single element using a user-defined `ReduceFunction`.
-For each group of input elements, a `ReduceFunction` successively combines pairs of elements into one element until only a single element for each group remains.
+A Reduce transformation that is applied on a grouped DataSet reduces each group to a single
+element using a user-defined reduce function.
+For each group of input elements, a reduce function successively combines pairs of elements into one
+element until only a single element for each group remains.
-#### Reduce on DataSet grouped by KeySelector Function
+#### Reduce on DataSet Grouped by KeySelector Function
-A `KeySelector` function extracts a key value from each element of a `DataSet`. The extracted key value is used to group the `DataSet`.
-The following code shows how to group a POJO `DataSet` using a `KeySelector` function and to reduce it with a `ReduceFunction`.
+A key-selector function extracts a key value from each element of a DataSet. The extracted key
+value is used to group the DataSet.
+The following code shows how to group a POJO DataSet using a key-selector function and to reduce it
+with a reduce function.
-```java
+
+
+
+~~~java
// some ordinary POJO
public class WC {
public String word;
@@ -162,14 +227,38 @@ DataSet wordCounts = words
})
// apply ReduceFunction on grouped DataSet
.reduce(new WordCounter());
-```
+~~~
+
+
+
+
+~~~scala
+// 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)
+}
+~~~
+
+
+
+
+
+~~~java
DataSet> tuples = // [...]
DataSet> reducedTuples =
tuples
@@ -177,19 +266,48 @@ DataSet> reducedTuples =
.groupBy(0,1)
// apply ReduceFunction on grouped DataSet
.reduce(new MyTupleReducer());
-```
+~~~
+
+
+
+
+~~~scala
+val tuples = DataSet[(String, Int, Double)] = // [...]
+// group on the first and second Tuple field
+val reducedTuples = tuples.groupBy(0, 1).reduce { ... }
+~~~
+
+
+#### Reduce on DataSet grouped by Case Class Fields
+
+When using Case Classes you can also specify the grouping key using the names of the fields:
-### GroupReduce on grouped DataSet
+~~~scala
+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 { ... }
+~~~
-A GroupReduce transformation that is applied on a grouped `DataSet` calls a user-defined `GroupReduceFunction` for each group. The difference
-between this and `Reduce` is that the user defined function gets the whole group at once.
-The function is invoked with an Iterable over all elements of a group and can return an arbitrary number of result elements using the collector.
+
+
+
+
+~~~java
public class DistinctReduce
implements GroupReduceFunction, Tuple2> {
@@ -217,20 +335,43 @@ DataSet> input = // [...]
DataSet> output = input
.groupBy(0) // group DataSet by the first tuple field
.reduceGroup(new DistinctReduce()); // apply GroupReduceFunction
-```
+~~~
+
+
+
+
+~~~scala
+val input: DataSet[(Int, String)] = // [...]
+val output = input.groupBy(0).reduceGroup {
+ (in, out: Collector[(Int, String)]) =>
+ in.toSet foreach (out.collect)
+ }
+~~~
+
+#### GroupReduce on DataSet Grouped by Case Class Fields
+
+Works analogous to grouping by Case Class fields in *Reduce* transformations.
-#### GroupReduce on DataSet grouped by KeySelector Function
-Works analogous to `KeySelector` functions in Reduce transformations.
+
+
+
+
+~~~java
// GroupReduceFunction that removes consecutive identical elements
public class DistinctReduce
implements GroupReduceFunction, Tuple2> {
@@ -259,19 +400,41 @@ DataSet output = input
.groupBy(0) // group DataSet by first field
.sortGroup(1, Order.ASCENDING) // sort groups on second tuple field
.reduceGroup(new DistinctReduce());
-```
+~~~
+
+
+
+
+~~~scala
+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)
+ }
+ }
+
+~~~
+
+
+
+
+
+~~~java
// Combinable GroupReduceFunction that computes two sums.
// Note that we use the RichGroupReduceFunction because it defines the combine method
@Combinable
@@ -302,9 +465,49 @@ public class MyCombinableGroupReducer
this.reduce(in, out);
}
}
-```
+~~~
+
+
+
+
+~~~scala
+
+// Combinable GroupReduceFunction that computes two sums.
+// Note that we use the RichGroupReduceFunction because it defines the combine method
+@Combinable
+class MyCombinableGroupReducer
+ extends RichGroupReduceFunction[(String, Int, Double), (String, Int, Double)] {}
+
+ def reduce(
+ in: java.lang.Iterable[(String, Int, Double)],
+ out: Collector[(String, Int, Double)]): Unit = {
+
+ val key: String = null
+ val intSum = 0
+ val doubleSum = 0.0
+
+ for (curr <- in) {
+ key = curr._1
+ intSum += curr._2
+ doubleSum += curr._3
+ }
+ // emit a tuple with both sums
+ out.collect(key, intSum, doubleSum);
+ }
+
+ def combine(
+ in: java.lang.Iterable[(String, Int, Double)],
+ out: Collector[(String, Int, Double)]): Unit = {
+ // in some cases combine() calls can simply be forwarded to reduce().
+ this.reduce(in, out)
+ }
+}
+~~~
+
+
+
+
+
+~~~java
DataSet> input = // [...]
DataSet> 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
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
-```java
+~~~java
// ReduceFunction that sums Integers
public class IntSummer implements ReduceFunction {
@Override
@@ -348,24 +567,47 @@ public class IntSummer implements ReduceFunction {
// [...]
DataSet intNumbers = // [...]
DataSet sum = intNumbers.reduce(new IntSummer());
-```
+~~~
+
+
+
+
+~~~scala
-Reducing a full `DataSet` using the Reduce transformation implies that the final Reduce operation cannot be done in parallel. However, a `ReduceFunction` is automatically combinable such that a Reduce transformation does not limit scalability for most use cases.
+~~~
+
+
+
+
+
+~~~java
DataSet input = // [...]
// apply a (preferably combinable) GroupReduceFunction to a DataSet
DataSet output = input.reduceGroup(new MyGroupReducer());
-```
+~~~
+
+
+
-**Note:** A GroupReduce transformation on a full `DataSet` cannot be done in parallel if the `GroupReduceFunction` is not combinable. Therefore, this can be a very compute intensive operation. See the paragraph on "Combineable `GroupReduceFunction`s" above to learn how to implement a combinable `GroupReduceFunction`.
+~~~scala
+
+~~~
+
+
+
+
-```java
+~~~java
DataSet> input = // [...]
DataSet> output = input
.aggregate(SUM, 0) // compute sum of the first field
.and(MIN, 1); // compute minimum of the second field
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
DataSet> input1 = // [...]
DataSet> input2 = // [...]
// result dataset is typed as Tuple2
@@ -411,16 +669,29 @@ DataSet, Tuple2>>
result = input1.join(input2)
.where(0) // key of the first input
.equalTo(1); // key of the second input
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
-```java
+~~~java
// some POJO
public class Rating {
public String name;
@@ -457,7 +728,17 @@ DataSet>
// applying the JoinFunction on joining pairs
.with(new PointWeighter());
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
DataSet> input1 = // [...]
DataSet> input2 = // [...]
DataSet
@@ -498,16 +782,29 @@ DataSet
// select and reorder fields of matching tuples
.projectFirst(0,2).projectSecond(1).projectFirst(1)
.types(Integer.class, String.class, Double.class, Byte.class);
-```
+~~~
+
+
+
-`projectFirst(int...)` and `projectSecond(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` `DataSet`s. In this case, `projectFirst()` or `projectSecond()` must be called without arguments to add a joined element to the output `Tuple`.
+~~~scala
+
+~~~
+
+
+
+
-```java
+~~~java
DataSet> input1 = // [...]
DataSet> input2 = // [...]
@@ -524,11 +821,21 @@ DataSet, Tuple2>>
input1.joinWithHuge(input2)
.where(0)
.equalTo(0);
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
public class Coord {
public int id;
public int x;
@@ -565,13 +875,26 @@ DataSet>
coords1.cross(coords2)
// apply CrossFunction
.with(new EuclideanDistComputer());
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
DataSet> input1 = // [...]
DataSet> input2 = // [...]
DataSet
@@ -580,15 +903,28 @@ DataSet
// select and reorder fields of matching tuples
.projectSecond(0).projectFirst(1,0).projectSecond(1)
.types(Integer.class, Byte.class, Integer.class, Double.class);
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
DataSet> input1 = // [...]
DataSet> input2 = // [...]
@@ -605,21 +941,34 @@ DataSet>
input1.crossWithHuge(input2)
// apply a projection (or any Cross function)
.projectFirst(0,1).projectSecond(1).types(Integer.class, String.class, String.class)
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
+
+
-```java
+~~~java
// Some CoGroupFunction definition
class MyCoGrouper
implements CoGroupFunction, Tuple2, Double> {
@@ -655,23 +1004,44 @@ DataSet output = iVals.coGroup(dVals)
.equalTo(0)
// apply CoGroup function on each pair of groups
.with(new MyCoGrouper());
-```
+~~~
-#### CoGroup on DataSets grouped by Key Selector Function
+
+
-Works analogous to key selector functions in Join transformations.
+~~~scala
+
+~~~
+
+
+
+
+
+~~~java
DataSet> vals1 = // [...]
DataSet> vals2 = // [...]
DataSet> vals3 = // [...]
DataSet> unioned = vals1.union(vals2)
.union(vals3);
-```
+~~~
+
+
+
+
+~~~scala
+
+~~~
+
+
-You can use the `Tuple` classes for composite types. Tuples contain a fix number of fields of
+You can use the Tuple classes for composite types. Tuples contain a fix number of fields of
various types. The Java API provides classes from `Tuple1` up to `Tuple25`. Every field of a tuple
can be an arbitrary Flink type - including further tuples, resulting in nested tuples. Fields of a
Tuple can be accessed directly using the fields `tuple.f4`, or using the generic getter method
@@ -1266,7 +1266,7 @@ wordCounts
{% endhighlight %}
In order to access fields more intuitively and to generate more readable code, it is also possible
-to extend a subclass of `Tuple`. You can add getters and setters with custom names that delegate to
+to extend a subclass of Tuple. You can add getters and setters with custom names that delegate to
the field positions. See this
{% gh_link /flink-examples/flink-java-examples/src/main/java/org/apache/flink/example/java/relational/TPCHQuery3.java "example" %} for an
illustration how to make use of that mechanism.