apache/incubator-flink — pull request #243 by he-sk (2014-11-28 14:46:47+00:00)
Title: [FLINK-1293] Add support for out-of-place aggregations

This patch adds support for multiple aggregations on the same field and aggregations which change the output type, e.g., count and average.

It adds the following aggregation syntax to the Java API:

// A string key and a long value
DataSet> ds = ...

// count all the elements
DataSet> result = ds.aggregate(count());

// count elements in each group
DataSet> result = ds.groupBy(0).aggregate(allKeys(), count());

// same as above but explicitly state the returned key(s)
DataSet> result = ds.groupBy(0).aggregate(key(0), count());

// only return counts, drop keys
DataSet> result = ds.groupBy(0).aggregate(count());

// average reuses count and sum
DataSet> result =
ds.groupBy(0).aggregate(allKeys(), count(), sum(1), average(1));

Five aggregation functions are supported: min, max, sum, count and average. Notice that count does not take a field reference in the example above.

Internally, the aggregation is implemented by the following operator chain:

Input -> Map1 -> Reduce -> Map2 -> Output

- Map1 constructs an intermediate tuple on which the aggregation is performed. It's main task is to copy tuple fields that are used in multiple aggregation functions and initialize fields required for the internal implementation of aggregation functions. For example, average requires a field to count the tuples.

- Reduce performs the actual aggregation

- Map2 computes the final result of the aggregation function, e.g., computing the average from the sum and the count, and drops the fields used in groupings that are not requested in the output.

Average is implemented to reuse an existing sum and/or count function. In the last example above, the intermediate tuple contains only 3 fields: the key, a field to compute the sum, and a field to compute the count. The field holding the average is added after the Reduce operator in the Map2 operator.

Currently, only the Java API is implemented. My Scala knowledge is fairly limited, so it would be great if somebody else could pick that up.

Also, the result type of aggregate is simply <T extends Tuple>. To support type inference at compile time, it is necessary to integrate the work by Chen (https://github.com/apache/incubator-flink/pull/194).