[FLINK-4017] [py] Add Aggregation support to Python API #2115
Conversation
Assembles and applies a GroupReduceFunction using pre-defined AggregationOperations in Python. References to aggregations in PythonOperationInfo and other Java classes in the Python API removed since aggregations are now handled by Python.
|
Thank you for your contribution, I will review this tomorrow :) |
| :param field: The index of the Tuple field on which to perform the function. | ||
| :return: A GroupReduceOperator that represents the aggregated DataSet. | ||
| """ | ||
| child_set = self.reduce_group(aggregation(field), combinable=True) |
There was a problem hiding this comment.
self.reduce_group(AggregationFunction(aggregation, field), combinable=True)
There was a problem hiding this comment.
there should also be a test case for a non-grouped aggregation.
|
Looks good overall. I found 1 big issue, the rest are minor things. |
|
Please write a comment when you update the PR, we don't get any notifications for pushed commits :) |
|
I'll have to try it out to be sure, but i can't find a problem looking through the code. |
| from flink.functions.Aggregation import Sum, Min | ||
|
|
||
| input = # [...] | ||
| output = input.aggregate(Sum, 0).agg_and(Min, 2) |
There was a problem hiding this comment.
documentation wasn't updated for the rename to and_agg. The same applies to the python.md
|
Found a small issue in the documentation, otherwise +1 to merge. |
|
I've addressed the documentation issue, thanks. |
|
merging |
Adds Aggregation support in the Python API accessible through
.aggregate()and.agg_and(). (I was unable to use.and()in Python because 'and' is a keyword.)