You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ji...@apache.org on 2019/07/02 00:01:27 UTC
[flink] branch master updated: [FLINK-11147][table][docs] Add
documentation for TableAggregate Function
This is an automated email from the ASF dual-hosted git repository.
jincheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6a12908 [FLINK-11147][table][docs] Add documentation for TableAggregate Function
6a12908 is described below
commit 6a12908b15c398e37f8603cd84e0d30e14d07784
Author: hequn8128 <ch...@gmail.com>
AuthorDate: Fri Jun 21 12:48:11 2019 +0800
[FLINK-11147][table][docs] Add documentation for TableAggregate Function
This close #8669
---
docs/dev/table/tableApi.md | 4 +-
docs/dev/table/tableApi.zh.md | 4 +-
docs/dev/table/udfs.md | 704 +++++++++++++++++++++++++++++++++++++++---
docs/dev/table/udfs.zh.md | 704 +++++++++++++++++++++++++++++++++++++++---
docs/fig/udtagg-mechanism.png | Bin 0 -> 150838 bytes
5 files changed, 1328 insertions(+), 88 deletions(-)
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index 4fb9f75..a0bd51a 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -2645,7 +2645,7 @@ Table table = input
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
- <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight java %}
/**
* Accumulator for Top2.
@@ -2850,7 +2850,7 @@ val table = input
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
- <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight scala %}
import java.lang.{Integer => JInteger}
import org.apache.flink.table.api.Types
diff --git a/docs/dev/table/tableApi.zh.md b/docs/dev/table/tableApi.zh.md
index 33ad6f9..7a75c0a 100644
--- a/docs/dev/table/tableApi.zh.md
+++ b/docs/dev/table/tableApi.zh.md
@@ -2644,7 +2644,7 @@ Table table = input
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
- <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight java %}
/**
* Accumulator for Top2.
@@ -2849,7 +2849,7 @@ val table = input
</td>
<td>
<p>Similar to a <b>GroupBy Aggregation</b>. Groups the rows on the grouping keys with the following running table aggregation operator to aggregate rows group-wise. The difference from an AggregateFunction is that TableAggregateFunction may return 0 or more records for a group. You have to close the "flatAggregate" with a select statement. And the select statement does not support aggregate functions.</p>
- <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
+ <p>Instead of using <code>emitValue</code> to output results, you can also use the <code>emitUpdateWithRetract</code> method. Different from <code>emitValue</code>, <code>emitUpdateWithRetract</code> is used to emit values that have been updated. This method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The <code>emitUpdateWithRetract</code> method will be used in preference to the <code> [...]
{% highlight scala %}
import java.lang.{Integer => JInteger}
import org.apache.flink.table.api.Types
diff --git a/docs/dev/table/udfs.md b/docs/dev/table/udfs.md
index c4179ee..87bc804 100644
--- a/docs/dev/table/udfs.md
+++ b/docs/dev/table/udfs.md
@@ -269,7 +269,7 @@ class CustomTypeSplit extends TableFunction[Row] {
Aggregation Functions
---------------------
-User-Defined Aggregate Functions (UDAGGs) aggregate a table (one ore more rows with one or more attributes) to a scalar value.
+User-Defined Aggregate Functions (UDAGGs) aggregate a table (one or more rows with one or more attributes) to a scalar value.
<center>
<img alt="UDAGG mechanism" src="{{ site.baseurl }}/fig/udagg-mechanism.png" width="80%">
@@ -305,23 +305,49 @@ Detailed documentation for all methods of `AggregateFunction` is given below.
<div data-lang="java" markdown="1">
{% highlight java %}
/**
- * Base class for aggregation functions.
+ * Base class for user-defined aggregates and table aggregates.
*
- * @param <T> the type of the aggregation result
+ * @param <T> the type of the aggregation result.
* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
* aggregated values which are needed to compute an aggregation result.
- * AggregateFunction represents its state using accumulator, thereby the state of the
- * AggregateFunction must be put into the accumulator.
*/
-public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
+public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {
/**
- * Creates and init the Accumulator for this [[AggregateFunction]].
+ * Creates and init the Accumulator for this (table)aggregate function.
*
* @return the accumulator with the initial value
*/
public ACC createAccumulator(); // MANDATORY
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ public TypeInformation<T> getResultType = null; // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
+}
+
+/**
+ * Base class for aggregation functions.
+ *
+ * @param <T> the type of the aggregation result
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ * AggregateFunction represents its state using accumulator, thereby the state of the
+ * AggregateFunction must be put into the accumulator.
+ */
+public abstract class AggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+
/** Processes the input values and update the provided accumulator instance. The method
* accumulate can be overloaded with different custom types and arguments. An AggregateFunction
* requires at least one accumulate() method.
@@ -350,7 +376,7 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* be noted that the accumulator may contain the previous aggregated
* results. Therefore user should not replace or clean this instance in the
* custom merge method.
- * @param its an [[java.lang.Iterable]] pointed to a group of accumulators that will be
+ * @param its an {@link java.lang.Iterable} pointed to a group of accumulators that will be
* merged.
*/
public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL
@@ -381,28 +407,45 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* @return true if the AggregateFunction requires an OVER window, false otherwise.
*/
public Boolean requiresOver = false; // PRE-DEFINED
+}
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @tparam T the type of the aggregation result.
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+abstract class UserDefinedAggregateFunction[T, ACC] extends UserDefinedFunction {
/**
- * Returns the TypeInformation of the AggregateFunction's result.
+ * Creates and init the Accumulator for this (table)aggregate function.
*
- * @return The TypeInformation of the AggregateFunction's result or null if the result type
- * should be automatically inferred.
+ * @return the accumulator with the initial value
*/
- public TypeInformation<T> getResultType = null; // PRE-DEFINED
+ def createAccumulator(): ACC // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ def getResultType: TypeInformation[T] = null // PRE-DEFINED
/**
- * Returns the TypeInformation of the AggregateFunction's accumulator.
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
*
- * @return The TypeInformation of the AggregateFunction's accumulator or null if the
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
* accumulator type should be automatically inferred.
*/
- public TypeInformation<T> getAccumulatorType = null; // PRE-DEFINED
+ def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
}
-{% endhighlight %}
-</div>
-<div data-lang="scala" markdown="1">
-{% highlight scala %}
/**
* Base class for aggregation functions.
*
@@ -412,13 +455,7 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* AggregateFunction represents its state using accumulator, thereby the state of the
* AggregateFunction must be put into the accumulator.
*/
-abstract class AggregateFunction[T, ACC] extends UserDefinedFunction {
- /**
- * Creates and init the Accumulator for this [[AggregateFunction]].
- *
- * @return the accumulator with the initial value
- */
- def createAccumulator(): ACC // MANDATORY
+abstract class AggregateFunction[T, ACC] extends UserDefinedAggregateFunction[T, ACC] {
/**
* Processes the input values and update the provided accumulator instance. The method
@@ -480,22 +517,6 @@ abstract class AggregateFunction[T, ACC] extends UserDefinedFunction {
* @return true if the AggregateFunction requires an OVER window, false otherwise.
*/
def requiresOver: Boolean = false // PRE-DEFINED
-
- /**
- * Returns the TypeInformation of the AggregateFunction's result.
- *
- * @return The TypeInformation of the AggregateFunction's result or null if the result type
- * should be automatically inferred.
- */
- def getResultType: TypeInformation[T] = null // PRE-DEFINED
-
- /**
- * Returns the TypeInformation of the AggregateFunction's accumulator.
- *
- * @return The TypeInformation of the AggregateFunction's accumulator or null if the
- * accumulator type should be automatically inferred.
- */
- def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
}
{% endhighlight %}
</div>
@@ -655,6 +676,605 @@ tEnv.sqlQuery("SELECT user, wAvg(points, level) AS avgPoints FROM userScores GRO
{% top %}
+Table Aggregation Functions
+---------------------
+
+User-Defined Table Aggregate Functions (UDTAGGs) aggregate a table (one or more rows with one or more attributes) to a result table with multi rows and columns.
+
+<center>
+<img alt="UDAGG mechanism" src="{{ site.baseurl }}/fig/udtagg-mechanism.png" width="80%">
+</center>
+
+The above figure shows an example of a table aggregation. Assume you have a table that contains data about beverages. The table consists of three columns, `id`, `name` and `price` and 5 rows. Imagine you need to find the top 2 highest prices of all beverages in the table, i.e., perform a `top2()` table aggregation. You would need to check each of the 5 rows and the result would be a table with the top 2 values.
+
+User-defined table aggregation functions are implemented by extending the `TableAggregateFunction` class. A `TableAggregateFunction` works as follows. First, it needs an `accumulator`, which is the data structure that holds the intermediate result of the aggregation. An empty accumulator is created by calling the `createAccumulator()` method of the `TableAggregateFunction`. Subsequently, the `accumulate()` method of the function is called for each input row to update the accumulator. Onc [...]
+
+**The following methods are mandatory for each `TableAggregateFunction`:**
+
+- `createAccumulator()`
+- `accumulate()`
+
+Flink’s type extraction facilities can fail to identify complex data types, e.g., if they are not basic types or simple POJOs. So similar to `ScalarFunction` and `TableFunction`, `TableAggregateFunction` provides methods to specify the `TypeInformation` of the result type (through
+ `TableAggregateFunction#getResultType()`) and the type of the accumulator (through `TableAggregateFunction#getAccumulatorType()`).
+
+Besides the above methods, there are a few contracted methods that can be
+optionally implemented. While some of these methods allow the system more efficient query execution, others are mandatory for certain use cases. For instance, the `merge()` method is mandatory if the aggregation function should be applied in the context of a session group window (the accumulators of two session windows need to be joined when a row is observed that "connects" them).
+
+**The following methods of `TableAggregateFunction` are required depending on the use case:**
+
+- `retract()` is required for aggregations on bounded `OVER` windows.
+- `merge()` is required for many batch aggregations and session window aggregations.
+- `resetAccumulator()` is required for many batch aggregations.
+- `emitValue()` is required for batch and window aggregations.
+
+**The following methods of `TableAggregateFunction` are used to improve the performance of streaming jobs:**
+
+- `emitUpdateWithRetract()` is used to emit values that have been updated under retract mode.
+
+For `emitValue` method, it emits full data according to the accumulator. Take TopN as an example, `emitValue` emit all top n values each time. This may bring performance problems for streaming jobs. To improve the performance, a user can also implement `emitUpdateWithRetract` method to improve the performance. The method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The method will be used in pre [...]
+
+All methods of `TableAggregateFunction` must be declared as `public`, not `static` and named exactly as the names mentioned above. The methods `createAccumulator`, `getResultType`, and `getAccumulatorType` are defined in the parent abstract class of `TableAggregateFunction`, while others are contracted methods. In order to define a table aggregate function, one has to extend the base class `org.apache.flink.table.functions.TableAggregateFunction` and implement one (or more) `accumulate` [...]
+
+Detailed documentation for all methods of `TableAggregateFunction` is given below.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @param <T> the type of the aggregation result.
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {
+
+ /**
+ * Creates and init the Accumulator for this (table)aggregate function.
+ *
+ * @return the accumulator with the initial value
+ */
+ public ACC createAccumulator(); // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ public TypeInformation<T> getResultType = null; // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
+}
+
+/**
+ * Base class for table aggregation functions.
+ *
+ * @param <T> the type of the aggregation result
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute a table aggregation result.
+ * TableAggregateFunction represents its state using accumulator, thereby the state of
+ * the TableAggregateFunction must be put into the accumulator.
+ */
+public abstract class TableAggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+
+ /** Processes the input values and update the provided accumulator instance. The method
+ * accumulate can be overloaded with different custom types and arguments. A TableAggregateFunction
+ * requires at least one accumulate() method.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ public void accumulate(ACC accumulator, [user defined inputs]); // MANDATORY
+
+ /**
+ * Retracts the input values from the accumulator instance. The current design assumes the
+ * inputs are the values that have been previously accumulated. The method retract can be
+ * overloaded with different custom types and arguments. This function must be implemented for
+ * datastream bounded over aggregate.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ public void retract(ACC accumulator, [user defined inputs]); // OPTIONAL
+
+ /**
+ * Merges a group of accumulator instances into one accumulator instance. This function must be
+ * implemented for datastream session window grouping aggregate and dataset grouping aggregate.
+ *
+ * @param accumulator the accumulator which will keep the merged aggregate results. It should
+ * be noted that the accumulator may contain the previous aggregated
+ * results. Therefore user should not replace or clean this instance in the
+ * custom merge method.
+ * @param its an {@link java.lang.Iterable} pointed to a group of accumulators that will be
+ * merged.
+ */
+ public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the collector used to output data
+ */
+ public void emitValue(ACC accumulator, Collector<T> out); // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated.
+ * This method outputs data incrementally in retract mode, i.e., once there is an update, we
+ * have to retract old records before sending new updated ones. The emitUpdateWithRetract
+ * method will be used in preference to the emitValue method if both methods are defined in the
+ * table aggregate function, because the method is treated to be more efficient than emitValue
+ * as it can outputvalues incrementally.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the retractable collector used to output data. Use collect method
+ * to output(add) records and use retract method to retract(delete)
+ * records.
+ */
+ public void emitUpdateWithRetract(ACC accumulator, RetractableCollector<T> out); // OPTIONAL
+
+ /**
+ * Collects a record and forwards it. The collector can output retract messages with the retract
+ * method. Note: only use it in {@code emitRetractValueIncrementally}.
+ */
+ public interface RetractableCollector<T> extends Collector<T> {
+
+ /**
+ * Retract a record.
+ *
+ * @param record The record to retract.
+ */
+ void retract(T record);
+ }
+}
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @tparam T the type of the aggregation result.
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+abstract class UserDefinedAggregateFunction[T, ACC] extends UserDefinedFunction {
+
+ /**
+ * Creates and init the Accumulator for this (table)aggregate function.
+ *
+ * @return the accumulator with the initial value
+ */
+ def createAccumulator(): ACC // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ def getResultType: TypeInformation[T] = null // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
+}
+
+/**
+ * Base class for table aggregation functions.
+ *
+ * @tparam T the type of the aggregation result
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ * TableAggregateFunction represents its state using accumulator, thereby the state of
+ * the TableAggregateFunction must be put into the accumulator.
+ */
+abstract class TableAggregateFunction[T, ACC] extends UserDefinedAggregateFunction[T, ACC] {
+
+ /**
+ * Processes the input values and update the provided accumulator instance. The method
+ * accumulate can be overloaded with different custom types and arguments. A TableAggregateFunction
+ * requires at least one accumulate() method.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ def accumulate(accumulator: ACC, [user defined inputs]): Unit // MANDATORY
+
+ /**
+ * Retracts the input values from the accumulator instance. The current design assumes the
+ * inputs are the values that have been previously accumulated. The method retract can be
+ * overloaded with different custom types and arguments. This function must be implemented for
+ * datastream bounded over aggregate.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ def retract(accumulator: ACC, [user defined inputs]): Unit // OPTIONAL
+
+ /**
+ * Merges a group of accumulator instances into one accumulator instance. This function must be
+ * implemented for datastream session window grouping aggregate and dataset grouping aggregate.
+ *
+ * @param accumulator the accumulator which will keep the merged aggregate results. It should
+ * be noted that the accumulator may contain the previous aggregated
+ * results. Therefore user should not replace or clean this instance in the
+ * custom merge method.
+ * @param its an [[java.lang.Iterable]] pointed to a group of accumulators that will be
+ * merged.
+ */
+ def merge(accumulator: ACC, its: java.lang.Iterable[ACC]): Unit // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the collector used to output data
+ */
+ def emitValue(accumulator: ACC, out: Collector[T]): Unit // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated.
+ * This method outputs data incrementally in retract mode, i.e., once there is an update, we
+ * have to retract old records before sending new updated ones. The emitUpdateWithRetract
+ * method will be used in preference to the emitValue method if both methods are defined in the
+ * table aggregate function, because the method is treated to be more efficient than emitValue
+ * as it can outputvalues incrementally.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the retractable collector used to output data. Use collect method
+ * to output(add) records and use retract method to retract(delete)
+ * records.
+ */
+ def emitUpdateWithRetract(accumulator: ACC, out: RetractableCollector[T]): Unit // OPTIONAL
+
+ /**
+ * Collects a record and forwards it. The collector can output retract messages with the retract
+ * method. Note: only use it in `emitRetractValueIncrementally`.
+ */
+ trait RetractableCollector[T] extends Collector[T] {
+
+ /**
+ * Retract a record.
+ *
+ * @param record The record to retract.
+ */
+ def retract(record: T): Unit
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+
+The following example shows how to
+
+- define a `TableAggregateFunction` that calculates the top 2 values on a given column,
+- register the function in the `TableEnvironment`, and
+- use the function in a Table API query(TableAggregateFunction is only supported by Table API).
+
+To calculate the top 2 values, the accumulator needs to store the biggest 2 values of all the data that has been accumulated. In our example we define a class `Top2Accum` to be the accumulator. Accumulators are automatically backup-ed by Flink's checkpointing mechanism and restored in case of a failure to ensure exactly-once semantics.
+
+The `accumulate()` method of our `Top2` `TableAggregateFunction` has two inputs. The first one is the `Top2Accum` accumulator, the other one is the user-defined input: input value `v`. Although the `merge()` method is not mandatory for most table aggregation types, we provide it below as examples. Please note that we used Java primitive types and defined `getResultType()` and `getAccumulatorType()` methods in the Scala example because Flink type extraction does not work very well for Sca [...]
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+/**
+ * Accumulator for Top2.
+ */
+public class Top2Accum {
+ public Integer first;
+ public Integer second;
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {
+
+ @Override
+ public Top2Accum createAccumulator() {
+ Top2Accum acc = new Top2Accum();
+ acc.first = Integer.MIN_VALUE;
+ acc.second = Integer.MIN_VALUE;
+ return acc;
+ }
+
+
+ public void accumulate(Top2Accum acc, Integer v) {
+ if (v > acc.first) {
+ acc.second = acc.first;
+ acc.first = v;
+ } else if (v > acc.second) {
+ acc.second = v;
+ }
+ }
+
+ public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {
+ for (Top2Accum otherAcc : iterable) {
+ accumulate(acc, otherAcc.first);
+ accumulate(acc, otherAcc.second);
+ }
+ }
+
+ public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {
+ // emit the value and rank
+ if (acc.first != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.first, 1));
+ }
+ if (acc.second != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.second, 2));
+ }
+ }
+}
+
+// register function
+StreamTableEnvironment tEnv = ...
+tEnv.registerFunction("top2", new Top2());
+
+// init table
+Table tab = ...;
+
+// use function
+tab.groupBy("key")
+ .flatAggregate("top2(a) as (v, rank)")
+ .select("key, v, rank");
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import java.lang.{Integer => JInteger}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.functions.TableAggregateFunction
+
+/**
+ * Accumulator for top2.
+ */
+class Top2Accum {
+ var first: JInteger = _
+ var second: JInteger = _
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {
+
+ override def createAccumulator(): Top2Accum = {
+ val acc = new Top2Accum
+ acc.first = Int.MinValue
+ acc.second = Int.MinValue
+ acc
+ }
+
+ def accumulate(acc: Top2Accum, v: Int) {
+ if (v > acc.first) {
+ acc.second = acc.first
+ acc.first = v
+ } else if (v > acc.second) {
+ acc.second = v
+ }
+ }
+
+ def merge(acc: Top2Accum, its: JIterable[Top2Accum]): Unit = {
+ val iter = its.iterator()
+ while (iter.hasNext) {
+ val top2 = iter.next()
+ accumulate(acc, top2.first)
+ accumulate(acc, top2.second)
+ }
+ }
+
+ def emitValue(acc: Top2Accum, out: Collector[JTuple2[JInteger, JInteger]]): Unit = {
+ // emit the value and rank
+ if (acc.first != Int.MinValue) {
+ out.collect(JTuple2.of(acc.first, 1))
+ }
+ if (acc.second != Int.MinValue) {
+ out.collect(JTuple2.of(acc.second, 2))
+ }
+ }
+}
+
+// init table
+val tab = ...
+
+// use function
+tab
+ .groupBy('key)
+ .flatAggregate(top2('a) as ('v, 'rank))
+ .select('key, 'v, 'rank)
+
+{% endhighlight %}
+</div>
+</div>
+
+
+The following example shows how to use `emitUpdateWithRetract` method to emit only updates. To emit only updates, in our example, the accumulator keeps both old and new top 2 values. Note: if the N of topN is big, it may inefficient to keep both old and new values. One way to solve this case is to store the input record into the accumulator in `accumulate` method and then perform calculation in `emitUpdateWithRetract`.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+/**
+ * Accumulator for Top2.
+ */
+public class Top2Accum {
+ public Integer first;
+ public Integer second;
+ public Integer oldFirst;
+ public Integer oldSecond;
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {
+
+ @Override
+ public Top2Accum createAccumulator() {
+ Top2Accum acc = new Top2Accum();
+ acc.first = Integer.MIN_VALUE;
+ acc.second = Integer.MIN_VALUE;
+ acc.oldFirst = Integer.MIN_VALUE;
+ acc.oldSecond = Integer.MIN_VALUE;
+ return acc;
+ }
+
+ public void accumulate(Top2Accum acc, Integer v) {
+ if (v > acc.first) {
+ acc.second = acc.first;
+ acc.first = v;
+ } else if (v > acc.second) {
+ acc.second = v;
+ }
+ }
+
+ public void emitUpdateWithRetract(Top2Accum acc, RetractableCollector<Tuple2<Integer, Integer>> out) {
+ if (!acc.first.equals(acc.oldFirst)) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldFirst != Integer.MIN_VALUE) {
+ out.retract(Tuple2.of(acc.oldFirst, 1));
+ }
+ out.collect(Tuple2.of(acc.first, 1));
+ acc.oldFirst = acc.first;
+ }
+
+ if (!acc.second.equals(acc.oldSecond)) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldSecond != Integer.MIN_VALUE) {
+ out.retract(Tuple2.of(acc.oldSecond, 2));
+ }
+ out.collect(Tuple2.of(acc.second, 2));
+ acc.oldSecond = acc.second;
+ }
+ }
+}
+
+// register function
+StreamTableEnvironment tEnv = ...
+tEnv.registerFunction("top2", new Top2());
+
+// init table
+Table tab = ...;
+
+// use function
+tab.groupBy("key")
+ .flatAggregate("top2(a) as (v, rank)")
+ .select("key, v, rank");
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import java.lang.{Integer => JInteger}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.functions.TableAggregateFunction
+
+/**
+ * Accumulator for top2.
+ */
+class Top2Accum {
+ var first: JInteger = _
+ var second: JInteger = _
+ var oldFirst: JInteger = _
+ var oldSecond: JInteger = _
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {
+
+ override def createAccumulator(): Top2Accum = {
+ val acc = new Top2Accum
+ acc.first = Int.MinValue
+ acc.second = Int.MinValue
+ acc.oldFirst = Int.MinValue
+ acc.oldSecond = Int.MinValue
+ acc
+ }
+
+ def accumulate(acc: Top2Accum, v: Int) {
+ if (v > acc.first) {
+ acc.second = acc.first
+ acc.first = v
+ } else if (v > acc.second) {
+ acc.second = v
+ }
+ }
+
+ def emitUpdateWithRetract(
+ acc: Top2Accum,
+ out: RetractableCollector[JTuple2[JInteger, JInteger]])
+ : Unit = {
+ if (acc.first != acc.oldFirst) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldFirst != Int.MinValue) {
+ out.retract(JTuple2.of(acc.oldFirst, 1))
+ }
+ out.collect(JTuple2.of(acc.first, 1))
+ acc.oldFirst = acc.first
+ }
+ if (acc.second != acc.oldSecond) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldSecond != Int.MinValue) {
+ out.retract(JTuple2.of(acc.oldSecond, 2))
+ }
+ out.collect(JTuple2.of(acc.second, 2))
+ acc.oldSecond = acc.second
+ }
+ }
+}
+
+// init table
+val tab = ...
+
+// use function
+tab
+ .groupBy('key)
+ .flatAggregate(top2('a) as ('v, 'rank))
+ .select('key, 'v, 'rank)
+
+{% endhighlight %}
+</div>
+</div>
+
+
+{% top %}
+
Best Practices for Implementing UDFs
------------------------------------
diff --git a/docs/dev/table/udfs.zh.md b/docs/dev/table/udfs.zh.md
index 3ce2f3f..724e5bb 100644
--- a/docs/dev/table/udfs.zh.md
+++ b/docs/dev/table/udfs.zh.md
@@ -269,7 +269,7 @@ class CustomTypeSplit extends TableFunction[Row] {
Aggregation Functions
---------------------
-User-Defined Aggregate Functions (UDAGGs) aggregate a table (one ore more rows with one or more attributes) to a scalar value.
+User-Defined Aggregate Functions (UDAGGs) aggregate a table (one or more rows with one or more attributes) to a scalar value.
<center>
<img alt="UDAGG mechanism" src="{{ site.baseurl }}/fig/udagg-mechanism.png" width="80%">
@@ -305,23 +305,49 @@ Detailed documentation for all methods of `AggregateFunction` is given below.
<div data-lang="java" markdown="1">
{% highlight java %}
/**
- * Base class for aggregation functions.
+ * Base class for user-defined aggregates and table aggregates.
*
- * @param <T> the type of the aggregation result
+ * @param <T> the type of the aggregation result.
* @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
* aggregated values which are needed to compute an aggregation result.
- * AggregateFunction represents its state using accumulator, thereby the state of the
- * AggregateFunction must be put into the accumulator.
*/
-public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
+public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {
/**
- * Creates and init the Accumulator for this [[AggregateFunction]].
+ * Creates and init the Accumulator for this (table)aggregate function.
*
* @return the accumulator with the initial value
*/
public ACC createAccumulator(); // MANDATORY
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ public TypeInformation<T> getResultType = null; // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
+}
+
+/**
+ * Base class for aggregation functions.
+ *
+ * @param <T> the type of the aggregation result
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ * AggregateFunction represents its state using accumulator, thereby the state of the
+ * AggregateFunction must be put into the accumulator.
+ */
+public abstract class AggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+
/** Processes the input values and update the provided accumulator instance. The method
* accumulate can be overloaded with different custom types and arguments. An AggregateFunction
* requires at least one accumulate() method.
@@ -350,7 +376,7 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* be noted that the accumulator may contain the previous aggregated
* results. Therefore user should not replace or clean this instance in the
* custom merge method.
- * @param its an [[java.lang.Iterable]] pointed to a group of accumulators that will be
+ * @param its an {@link java.lang.Iterable} pointed to a group of accumulators that will be
* merged.
*/
public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL
@@ -381,28 +407,45 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* @return true if the AggregateFunction requires an OVER window, false otherwise.
*/
public Boolean requiresOver = false; // PRE-DEFINED
+}
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @tparam T the type of the aggregation result.
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+abstract class UserDefinedAggregateFunction[T, ACC] extends UserDefinedFunction {
/**
- * Returns the TypeInformation of the AggregateFunction's result.
+ * Creates and init the Accumulator for this (table)aggregate function.
*
- * @return The TypeInformation of the AggregateFunction's result or null if the result type
- * should be automatically inferred.
+ * @return the accumulator with the initial value
*/
- public TypeInformation<T> getResultType = null; // PRE-DEFINED
+ def createAccumulator(): ACC // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ def getResultType: TypeInformation[T] = null // PRE-DEFINED
/**
- * Returns the TypeInformation of the AggregateFunction's accumulator.
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
*
- * @return The TypeInformation of the AggregateFunction's accumulator or null if the
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
* accumulator type should be automatically inferred.
*/
- public TypeInformation<T> getAccumulatorType = null; // PRE-DEFINED
+ def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
}
-{% endhighlight %}
-</div>
-<div data-lang="scala" markdown="1">
-{% highlight scala %}
/**
* Base class for aggregation functions.
*
@@ -412,13 +455,7 @@ public abstract class AggregateFunction<T, ACC> extends UserDefinedFunction {
* AggregateFunction represents its state using accumulator, thereby the state of the
* AggregateFunction must be put into the accumulator.
*/
-abstract class AggregateFunction[T, ACC] extends UserDefinedFunction {
- /**
- * Creates and init the Accumulator for this [[AggregateFunction]].
- *
- * @return the accumulator with the initial value
- */
- def createAccumulator(): ACC // MANDATORY
+abstract class AggregateFunction[T, ACC] extends UserDefinedAggregateFunction[T, ACC] {
/**
* Processes the input values and update the provided accumulator instance. The method
@@ -480,22 +517,6 @@ abstract class AggregateFunction[T, ACC] extends UserDefinedFunction {
* @return true if the AggregateFunction requires an OVER window, false otherwise.
*/
def requiresOver: Boolean = false // PRE-DEFINED
-
- /**
- * Returns the TypeInformation of the AggregateFunction's result.
- *
- * @return The TypeInformation of the AggregateFunction's result or null if the result type
- * should be automatically inferred.
- */
- def getResultType: TypeInformation[T] = null // PRE-DEFINED
-
- /**
- * Returns the TypeInformation of the AggregateFunction's accumulator.
- *
- * @return The TypeInformation of the AggregateFunction's accumulator or null if the
- * accumulator type should be automatically inferred.
- */
- def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
}
{% endhighlight %}
</div>
@@ -655,6 +676,605 @@ tEnv.sqlQuery("SELECT user, wAvg(points, level) AS avgPoints FROM userScores GRO
{% top %}
+Table Aggregation Functions
+---------------------
+
+User-Defined Table Aggregate Functions (UDTAGGs) aggregate a table (one or more rows with one or more attributes) to a result table with multi rows and columns.
+
+<center>
+<img alt="UDAGG mechanism" src="{{ site.baseurl }}/fig/udtagg-mechanism.png" width="80%">
+</center>
+
+The above figure shows an example of a table aggregation. Assume you have a table that contains data about beverages. The table consists of three columns, `id`, `name` and `price` and 5 rows. Imagine you need to find the top 2 highest prices of all beverages in the table, i.e., perform a `top2()` table aggregation. You would need to check each of the 5 rows and the result would be a table with the top 2 values.
+
+User-defined table aggregation functions are implemented by extending the `TableAggregateFunction` class. A `TableAggregateFunction` works as follows. First, it needs an `accumulator`, which is the data structure that holds the intermediate result of the aggregation. An empty accumulator is created by calling the `createAccumulator()` method of the `TableAggregateFunction`. Subsequently, the `accumulate()` method of the function is called for each input row to update the accumulator. Onc [...]
+
+**The following methods are mandatory for each `TableAggregateFunction`:**
+
+- `createAccumulator()`
+- `accumulate()`
+
+Flink’s type extraction facilities can fail to identify complex data types, e.g., if they are not basic types or simple POJOs. So similar to `ScalarFunction` and `TableFunction`, `TableAggregateFunction` provides methods to specify the `TypeInformation` of the result type (through
+ `TableAggregateFunction#getResultType()`) and the type of the accumulator (through `TableAggregateFunction#getAccumulatorType()`).
+
+Besides the above methods, there are a few contracted methods that can be
+optionally implemented. While some of these methods allow the system more efficient query execution, others are mandatory for certain use cases. For instance, the `merge()` method is mandatory if the aggregation function should be applied in the context of a session group window (the accumulators of two session windows need to be joined when a row is observed that "connects" them).
+
+**The following methods of `TableAggregateFunction` are required depending on the use case:**
+
+- `retract()` is required for aggregations on bounded `OVER` windows.
+- `merge()` is required for many batch aggregations and session window aggregations.
+- `resetAccumulator()` is required for many batch aggregations.
+- `emitValue()` is required for batch and window aggregations.
+
+**The following methods of `TableAggregateFunction` are used to improve the performance of streaming jobs:**
+
+- `emitUpdateWithRetract()` is used to emit values that have been updated under retract mode.
+
+For `emitValue` method, it emits full data according to the accumulator. Take TopN as an example, `emitValue` emit all top n values each time. This may bring performance problems for streaming jobs. To improve the performance, a user can also implement `emitUpdateWithRetract` method to improve the performance. The method outputs data incrementally in retract mode, i.e., once there is an update, we have to retract old records before sending new updated ones. The method will be used in pre [...]
+
+All methods of `TableAggregateFunction` must be declared as `public`, not `static` and named exactly as the names mentioned above. The methods `createAccumulator`, `getResultType`, and `getAccumulatorType` are defined in the parent abstract class of `TableAggregateFunction`, while others are contracted methods. In order to define a table aggregate function, one has to extend the base class `org.apache.flink.table.functions.TableAggregateFunction` and implement one (or more) `accumulate` [...]
+
+Detailed documentation for all methods of `TableAggregateFunction` is given below.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @param <T> the type of the aggregation result.
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+public abstract class UserDefinedAggregateFunction<T, ACC> extends UserDefinedFunction {
+
+ /**
+ * Creates and init the Accumulator for this (table)aggregate function.
+ *
+ * @return the accumulator with the initial value
+ */
+ public ACC createAccumulator(); // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ public TypeInformation<T> getResultType = null; // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ public TypeInformation<ACC> getAccumulatorType = null; // PRE-DEFINED
+}
+
+/**
+ * Base class for table aggregation functions.
+ *
+ * @param <T> the type of the aggregation result
+ * @param <ACC> the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute a table aggregation result.
+ * TableAggregateFunction represents its state using accumulator, thereby the state of
+ * the TableAggregateFunction must be put into the accumulator.
+ */
+public abstract class TableAggregateFunction<T, ACC> extends UserDefinedAggregateFunction<T, ACC> {
+
+ /** Processes the input values and update the provided accumulator instance. The method
+ * accumulate can be overloaded with different custom types and arguments. A TableAggregateFunction
+ * requires at least one accumulate() method.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ public void accumulate(ACC accumulator, [user defined inputs]); // MANDATORY
+
+ /**
+ * Retracts the input values from the accumulator instance. The current design assumes the
+ * inputs are the values that have been previously accumulated. The method retract can be
+ * overloaded with different custom types and arguments. This function must be implemented for
+ * datastream bounded over aggregate.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ public void retract(ACC accumulator, [user defined inputs]); // OPTIONAL
+
+ /**
+ * Merges a group of accumulator instances into one accumulator instance. This function must be
+ * implemented for datastream session window grouping aggregate and dataset grouping aggregate.
+ *
+ * @param accumulator the accumulator which will keep the merged aggregate results. It should
+ * be noted that the accumulator may contain the previous aggregated
+ * results. Therefore user should not replace or clean this instance in the
+ * custom merge method.
+ * @param its an {@link java.lang.Iterable} pointed to a group of accumulators that will be
+ * merged.
+ */
+ public void merge(ACC accumulator, java.lang.Iterable<ACC> its); // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the collector used to output data
+ */
+ public void emitValue(ACC accumulator, Collector<T> out); // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated.
+ * This method outputs data incrementally in retract mode, i.e., once there is an update, we
+ * have to retract old records before sending new updated ones. The emitUpdateWithRetract
+ * method will be used in preference to the emitValue method if both methods are defined in the
+ * table aggregate function, because the method is treated to be more efficient than emitValue
+ * as it can outputvalues incrementally.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the retractable collector used to output data. Use collect method
+ * to output(add) records and use retract method to retract(delete)
+ * records.
+ */
+ public void emitUpdateWithRetract(ACC accumulator, RetractableCollector<T> out); // OPTIONAL
+
+ /**
+ * Collects a record and forwards it. The collector can output retract messages with the retract
+ * method. Note: only use it in {@code emitRetractValueIncrementally}.
+ */
+ public interface RetractableCollector<T> extends Collector<T> {
+
+ /**
+ * Retract a record.
+ *
+ * @param record The record to retract.
+ */
+ void retract(T record);
+ }
+}
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+/**
+ * Base class for user-defined aggregates and table aggregates.
+ *
+ * @tparam T the type of the aggregation result.
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ */
+abstract class UserDefinedAggregateFunction[T, ACC] extends UserDefinedFunction {
+
+ /**
+ * Creates and init the Accumulator for this (table)aggregate function.
+ *
+ * @return the accumulator with the initial value
+ */
+ def createAccumulator(): ACC // MANDATORY
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's result.
+ *
+ * @return The TypeInformation of the (table)aggregate function's result or null if the result
+ * type should be automatically inferred.
+ */
+ def getResultType: TypeInformation[T] = null // PRE-DEFINED
+
+ /**
+ * Returns the TypeInformation of the (table)aggregate function's accumulator.
+ *
+ * @return The TypeInformation of the (table)aggregate function's accumulator or null if the
+ * accumulator type should be automatically inferred.
+ */
+ def getAccumulatorType: TypeInformation[ACC] = null // PRE-DEFINED
+}
+
+/**
+ * Base class for table aggregation functions.
+ *
+ * @tparam T the type of the aggregation result
+ * @tparam ACC the type of the aggregation accumulator. The accumulator is used to keep the
+ * aggregated values which are needed to compute an aggregation result.
+ * TableAggregateFunction represents its state using accumulator, thereby the state of
+ * the TableAggregateFunction must be put into the accumulator.
+ */
+abstract class TableAggregateFunction[T, ACC] extends UserDefinedAggregateFunction[T, ACC] {
+
+ /**
+ * Processes the input values and update the provided accumulator instance. The method
+ * accumulate can be overloaded with different custom types and arguments. A TableAggregateFunction
+ * requires at least one accumulate() method.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ def accumulate(accumulator: ACC, [user defined inputs]): Unit // MANDATORY
+
+ /**
+ * Retracts the input values from the accumulator instance. The current design assumes the
+ * inputs are the values that have been previously accumulated. The method retract can be
+ * overloaded with different custom types and arguments. This function must be implemented for
+ * datastream bounded over aggregate.
+ *
+ * @param accumulator the accumulator which contains the current aggregated results
+ * @param [user defined inputs] the input value (usually obtained from a new arrived data).
+ */
+ def retract(accumulator: ACC, [user defined inputs]): Unit // OPTIONAL
+
+ /**
+ * Merges a group of accumulator instances into one accumulator instance. This function must be
+ * implemented for datastream session window grouping aggregate and dataset grouping aggregate.
+ *
+ * @param accumulator the accumulator which will keep the merged aggregate results. It should
+ * be noted that the accumulator may contain the previous aggregated
+ * results. Therefore user should not replace or clean this instance in the
+ * custom merge method.
+ * @param its an [[java.lang.Iterable]] pointed to a group of accumulators that will be
+ * merged.
+ */
+ def merge(accumulator: ACC, its: java.lang.Iterable[ACC]): Unit // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the collector used to output data
+ */
+ def emitValue(accumulator: ACC, out: Collector[T]): Unit // OPTIONAL
+
+ /**
+ * Called every time when an aggregation result should be materialized. The returned value
+ * could be either an early and incomplete result (periodically emitted as data arrive) or
+ * the final result of the aggregation.
+ *
+ * Different from emitValue, emitUpdateWithRetract is used to emit values that have been updated.
+ * This method outputs data incrementally in retract mode, i.e., once there is an update, we
+ * have to retract old records before sending new updated ones. The emitUpdateWithRetract
+ * method will be used in preference to the emitValue method if both methods are defined in the
+ * table aggregate function, because the method is treated to be more efficient than emitValue
+ * as it can outputvalues incrementally.
+ *
+ * @param accumulator the accumulator which contains the current
+ * aggregated results
+ * @param out the retractable collector used to output data. Use collect method
+ * to output(add) records and use retract method to retract(delete)
+ * records.
+ */
+ def emitUpdateWithRetract(accumulator: ACC, out: RetractableCollector[T]): Unit // OPTIONAL
+
+ /**
+ * Collects a record and forwards it. The collector can output retract messages with the retract
+ * method. Note: only use it in `emitRetractValueIncrementally`.
+ */
+ trait RetractableCollector[T] extends Collector[T] {
+
+ /**
+ * Retract a record.
+ *
+ * @param record The record to retract.
+ */
+ def retract(record: T): Unit
+ }
+}
+{% endhighlight %}
+</div>
+</div>
+
+
+The following example shows how to
+
+- define a `TableAggregateFunction` that calculates the top 2 values on a given column,
+- register the function in the `TableEnvironment`, and
+- use the function in a Table API query(TableAggregateFunction is only supported by Table API).
+
+To calculate the top 2 values, the accumulator needs to store the biggest 2 values of all the data that has been accumulated. In our example we define a class `Top2Accum` to be the accumulator. Accumulators are automatically backup-ed by Flink's checkpointing mechanism and restored in case of a failure to ensure exactly-once semantics.
+
+The `accumulate()` method of our `Top2` `TableAggregateFunction` has two inputs. The first one is the `Top2Accum` accumulator, the other one is the user-defined input: input value `v`. Although the `merge()` method is not mandatory for most table aggregation types, we provide it below as examples. Please note that we used Java primitive types and defined `getResultType()` and `getAccumulatorType()` methods in the Scala example because Flink type extraction does not work very well for Sca [...]
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+/**
+ * Accumulator for Top2.
+ */
+public class Top2Accum {
+ public Integer first;
+ public Integer second;
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {
+
+ @Override
+ public Top2Accum createAccumulator() {
+ Top2Accum acc = new Top2Accum();
+ acc.first = Integer.MIN_VALUE;
+ acc.second = Integer.MIN_VALUE;
+ return acc;
+ }
+
+
+ public void accumulate(Top2Accum acc, Integer v) {
+ if (v > acc.first) {
+ acc.second = acc.first;
+ acc.first = v;
+ } else if (v > acc.second) {
+ acc.second = v;
+ }
+ }
+
+ public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {
+ for (Top2Accum otherAcc : iterable) {
+ accumulate(acc, otherAcc.first);
+ accumulate(acc, otherAcc.second);
+ }
+ }
+
+ public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {
+ // emit the value and rank
+ if (acc.first != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.first, 1));
+ }
+ if (acc.second != Integer.MIN_VALUE) {
+ out.collect(Tuple2.of(acc.second, 2));
+ }
+ }
+}
+
+// register function
+StreamTableEnvironment tEnv = ...
+tEnv.registerFunction("top2", new Top2());
+
+// init table
+Table tab = ...;
+
+// use function
+tab.groupBy("key")
+ .flatAggregate("top2(a) as (v, rank)")
+ .select("key, v, rank");
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import java.lang.{Integer => JInteger}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.functions.TableAggregateFunction
+
+/**
+ * Accumulator for top2.
+ */
+class Top2Accum {
+ var first: JInteger = _
+ var second: JInteger = _
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {
+
+ override def createAccumulator(): Top2Accum = {
+ val acc = new Top2Accum
+ acc.first = Int.MinValue
+ acc.second = Int.MinValue
+ acc
+ }
+
+ def accumulate(acc: Top2Accum, v: Int) {
+ if (v > acc.first) {
+ acc.second = acc.first
+ acc.first = v
+ } else if (v > acc.second) {
+ acc.second = v
+ }
+ }
+
+ def merge(acc: Top2Accum, its: JIterable[Top2Accum]): Unit = {
+ val iter = its.iterator()
+ while (iter.hasNext) {
+ val top2 = iter.next()
+ accumulate(acc, top2.first)
+ accumulate(acc, top2.second)
+ }
+ }
+
+ def emitValue(acc: Top2Accum, out: Collector[JTuple2[JInteger, JInteger]]): Unit = {
+ // emit the value and rank
+ if (acc.first != Int.MinValue) {
+ out.collect(JTuple2.of(acc.first, 1))
+ }
+ if (acc.second != Int.MinValue) {
+ out.collect(JTuple2.of(acc.second, 2))
+ }
+ }
+}
+
+// init table
+val tab = ...
+
+// use function
+tab
+ .groupBy('key)
+ .flatAggregate(top2('a) as ('v, 'rank))
+ .select('key, 'v, 'rank)
+
+{% endhighlight %}
+</div>
+</div>
+
+
+The following example shows how to use `emitUpdateWithRetract` method to emit only updates. To emit only updates, in our example, the accumulator keeps both old and new top 2 values. Note: if the N of topN is big, it may inefficient to keep both old and new values. One way to solve this case is to store the input record into the accumulator in `accumulate` method and then perform calculation in `emitUpdateWithRetract`.
+
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+/**
+ * Accumulator for Top2.
+ */
+public class Top2Accum {
+ public Integer first;
+ public Integer second;
+ public Integer oldFirst;
+ public Integer oldSecond;
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+public static class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {
+
+ @Override
+ public Top2Accum createAccumulator() {
+ Top2Accum acc = new Top2Accum();
+ acc.first = Integer.MIN_VALUE;
+ acc.second = Integer.MIN_VALUE;
+ acc.oldFirst = Integer.MIN_VALUE;
+ acc.oldSecond = Integer.MIN_VALUE;
+ return acc;
+ }
+
+ public void accumulate(Top2Accum acc, Integer v) {
+ if (v > acc.first) {
+ acc.second = acc.first;
+ acc.first = v;
+ } else if (v > acc.second) {
+ acc.second = v;
+ }
+ }
+
+ public void emitUpdateWithRetract(Top2Accum acc, RetractableCollector<Tuple2<Integer, Integer>> out) {
+ if (!acc.first.equals(acc.oldFirst)) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldFirst != Integer.MIN_VALUE) {
+ out.retract(Tuple2.of(acc.oldFirst, 1));
+ }
+ out.collect(Tuple2.of(acc.first, 1));
+ acc.oldFirst = acc.first;
+ }
+
+ if (!acc.second.equals(acc.oldSecond)) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldSecond != Integer.MIN_VALUE) {
+ out.retract(Tuple2.of(acc.oldSecond, 2));
+ }
+ out.collect(Tuple2.of(acc.second, 2));
+ acc.oldSecond = acc.second;
+ }
+ }
+}
+
+// register function
+StreamTableEnvironment tEnv = ...
+tEnv.registerFunction("top2", new Top2());
+
+// init table
+Table tab = ...;
+
+// use function
+tab.groupBy("key")
+ .flatAggregate("top2(a) as (v, rank)")
+ .select("key, v, rank");
+
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight scala %}
+import java.lang.{Integer => JInteger}
+import org.apache.flink.table.api.Types
+import org.apache.flink.table.functions.TableAggregateFunction
+
+/**
+ * Accumulator for top2.
+ */
+class Top2Accum {
+ var first: JInteger = _
+ var second: JInteger = _
+ var oldFirst: JInteger = _
+ var oldSecond: JInteger = _
+}
+
+/**
+ * The top2 user-defined table aggregate function.
+ */
+class Top2 extends TableAggregateFunction[JTuple2[JInteger, JInteger], Top2Accum] {
+
+ override def createAccumulator(): Top2Accum = {
+ val acc = new Top2Accum
+ acc.first = Int.MinValue
+ acc.second = Int.MinValue
+ acc.oldFirst = Int.MinValue
+ acc.oldSecond = Int.MinValue
+ acc
+ }
+
+ def accumulate(acc: Top2Accum, v: Int) {
+ if (v > acc.first) {
+ acc.second = acc.first
+ acc.first = v
+ } else if (v > acc.second) {
+ acc.second = v
+ }
+ }
+
+ def emitUpdateWithRetract(
+ acc: Top2Accum,
+ out: RetractableCollector[JTuple2[JInteger, JInteger]])
+ : Unit = {
+ if (acc.first != acc.oldFirst) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldFirst != Int.MinValue) {
+ out.retract(JTuple2.of(acc.oldFirst, 1))
+ }
+ out.collect(JTuple2.of(acc.first, 1))
+ acc.oldFirst = acc.first
+ }
+ if (acc.second != acc.oldSecond) {
+ // if there is an update, retract old value then emit new value.
+ if (acc.oldSecond != Int.MinValue) {
+ out.retract(JTuple2.of(acc.oldSecond, 2))
+ }
+ out.collect(JTuple2.of(acc.second, 2))
+ acc.oldSecond = acc.second
+ }
+ }
+}
+
+// init table
+val tab = ...
+
+// use function
+tab
+ .groupBy('key)
+ .flatAggregate(top2('a) as ('v, 'rank))
+ .select('key, 'v, 'rank)
+
+{% endhighlight %}
+</div>
+</div>
+
+
+{% top %}
+
Best Practices for Implementing UDFs
------------------------------------
diff --git a/docs/fig/udtagg-mechanism.png b/docs/fig/udtagg-mechanism.png
new file mode 100644
index 0000000..d5e3b71
Binary files /dev/null and b/docs/fig/udtagg-mechanism.png differ