auroflow commented on code in PR #28937:
URL: https://github.com/apache/flink/pull/28937#discussion_r3765751343
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -348,3 +433,75 @@ def collect(self) -> List[Row]:
"""
with self._table.execute().collect() as rows:
return list(rows)
+
+
+@PublicEvolving()
+class GroupedDataFrame:
+ """
+ A DataFrame grouped by one or more keys and ready for aggregation.
+
+ Instances are created by :meth:`DataFrame.group_by`.
+
+ .. versionadded:: 2.4.0
+ """
+
+ def __init__(self, dataframe: DataFrame, grouping_keys: List[Expression]):
+ self._dataframe = dataframe
+ self._grouping_keys = grouping_keys
+
+ @PublicEvolving()
+ def agg(self, *aggs: Expression, **named_aggs: Expression) -> DataFrame:
+ """
+ Aggregate the rows in each group.
+
+ Grouping keys are included first in their supplied order, followed by
positional
+ aggregation expressions and then named aggregations. Each named
aggregation is aliased to
+ its keyword name.
+
+ :param aggs: Aggregation expressions.
+ :param named_aggs: Aggregation expressions keyed by their result
column names.
+ :return: A DataFrame containing the grouping keys and aggregation
results.
+ :raises TypeError: If an aggregation is not an expression.
+ :raises ValueError: If no aggregations are provided.
+
+ Example::
+
+ >>> import pyflink.dataframe as pf
+ >>> df = pf.from_records([
+ ... ("engineering", 10),
+ ... ("engineering", 20),
+ ... ("sales", 5),
+ ... ], schema=["department", "amount"])
+ >>> totals = df.group_by("department").agg(
+ ... pf.col("amount").sum.alias("total_amount"),
+ ... row_count=pf.col("amount").count,
+ ... )
+ >>> # totals schema: [department: STRING, total_amount: BIGINT,
+ >>> # row_count: BIGINT NOT NULL]
+
+ .. versionadded:: 2.4.0
+ """
+ aggregations = _normalize_aggregations(aggs, named_aggs)
+ grouped_table = self._dataframe._table.group_by(*self._grouping_keys)
+ return DataFrame(grouped_table.select(*self._grouping_keys,
*aggregations))
+
+
+# ======================== Internal Helpers ========================
+
+
+def _normalize_aggregations(
+ aggs: Tuple[Expression, ...], named_aggs: Dict[str, Expression]
+) -> List[Expression]:
+ if not aggs and not named_aggs:
+ raise ValueError("agg() requires at least one aggregation")
+
+ aggregations: List[Expression] = []
+ for aggregation in aggs:
+ if not isinstance(aggregation, Expression):
+ raise TypeError("agg() aggregations must be expressions")
Review Comment:
Yes, I standardized it to "expressions."
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
