You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2019/05/15 13:58:08 UTC
[GitHub] [flink] aljoscha commented on a change in pull request #8401:
[FLINK-12407][python] Add all table operators align Java Table API.
aljoscha commented on a change in pull request #8401: [FLINK-12407][python] Add all table operators align Java Table API.
URL: https://github.com/apache/flink/pull/8401#discussion_r284269071
##########
File path: flink-python/pyflink/table/table.py
##########
@@ -106,6 +113,344 @@ def where(self, predicate):
"""
return Table(self._j_table.where(predicate))
+ def group_by(self, fields):
+ """
+ Groups the elements on some grouping keys. Use this before a selection with aggregations
+ to perform the aggregation on a per-group basis. Similar to a SQL GROUP BY statement.
+
+ Example:
+ ::
+ >>> tab.group_by("key").select("key, value.avg")
+
+ :param fields: Group keys.
+ :return: The grouped table.
+ """
+ return GroupedTable(self._j_table.groupBy(fields))
+
+ def distinct(self):
+ """
+ Removes duplicate values and returns onl
+ Example:
+ ::
+ >>> tab.select("key, value").distinct()
+
+ :return: Result table.
+ """
+ return Table(self._j_table.distinct())
+
+ def join(self, right, join_predicate=None):
+ """
+ Joins two :class:`Table`s. Similar to a SQL join. The fields of the two joined
+ operations must not overlap, use :func:`~pyflink.table.Table.alias` to rename fields if
+ necessary. You can use where and select clauses after a join to further specify the
+ behaviour of the join.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment` .
+
+ Example:
+ ::
+ >>> left.join(right).where("a = b && c > 3").select("a, b, d")
+ >>> left.join(right, "a = b")
+
+ :param right: Right table.
+ :param join_predicate: Optional, the join predicate expression string.
+ :return: Result table.
+ """
+ if join_predicate is not None:
+ return Table(self._j_table.join(right._j_table, join_predicate))
+ else:
+ return Table(self._j_table.join(right._j_table))
+
+ def left_outer_join(self, right, join_predicate=None):
+ """
+ Joins two :class:`Table`s. Similar to a SQL left outer join. The fields of the two joined
+ operations must not overlap, use :func:`~pyflink.table.Table.as_` to rename fields if
+ necessary.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment` and its
+ :class:`TableConfig` must have null check enabled (default).
+
+ Example:
+ ::
+ >>> left.left_outer_join(right).select("a, b, d")
+ >>> left.left_outer_join(right, "a = b").select("a, b, d")
+
+ :param right: Right table.
+ :param join_predicate: Optional, the join predicate expression string.
+ :return: Result table.
+ """
+ if join_predicate is None:
+ return Table(self._j_table.leftOuterJoin(right._j_table))
+ else:
+ return Table(self._j_table.leftOuterJoin(
+ right._j_table, join_predicate))
+
+ def right_outer_join(self, right, join_predicate):
+ """
+ Joins two :class:`Table`s. Similar to a SQL right outer join. The fields of the two joined
+ operations must not overlap, use :func:`~pyflink.table.Table.as_` to rename fields if
+ necessary.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment` and its
+ :class:`TableConfig` must have null check enabled (default).
+
+ Example:
+ ::
+ >>> left.right_outer_join(right, "a = b").select("a, b, d")
+
+ :param right: Right table.
+ :param join_predicate: The join predicate expression string.
+ :return: Result table.
+ """
+ return Table(self._j_table.rightOuterJoin(
+ right._j_table, join_predicate))
+
+ def full_outer_join(self, right, join_predicate):
+ """
+ Joins two :class:`Table`s. Similar to a SQL full outer join. The fields of the two joined
+ operations must not overlap, use :func:`~pyflink.table.Table.as_` to rename fields if
+ necessary.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment` and its
+ :class:`TableConfig` must have null check enabled (default).
+
+ Example:
+ ::
+ >>> left.full_outer_join(right, "a = b").select("a, b, d")
+
+ :param right: Right table.
+ :param join_predicate: The join predicate expression string.
+ :return: Result table.
+ """
+ return Table(self._j_table.fullOuterJoin(
+ right._j_table, join_predicate))
+
+ def minus(self, right):
+ """
+ Minus of two :class:`Table`s with duplicate records removed.
+ Similar to a SQL EXCEPT clause. Minus returns records from the left table that do not
+ exist in the right table. Duplicate records in the left table are returned
+ exactly once, i.e., duplicates are removed. Both tables must have identical field types.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.minus(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.minus(right._j_table))
+
+ def minus_all(self, right):
+ """
+ Minus of two :class:`Table`s. Similar to a SQL EXCEPT ALL.
+ Similar to a SQL EXCEPT ALL clause. MinusAll returns the records that do not exist in
+ the right table. A record that is present n times in the left table and m times
+ in the right table is returned (n - m) times, i.e., as many duplicates as are present
+ in the right table are removed. Both tables must have identical field types.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.minus_all(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.minusAll(right._j_table))
+
+ def union(self, right):
+ """
+ Unions two :class:`Table`s with duplicate records removed.
+ Similar to a SQL UNION. The fields of the two union operations must fully overlap.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.union(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.union(right._j_table))
+
+ def union_all(self, right):
+ """
+ Unions two :class:`Table`s. Similar to a SQL UNION ALL. The fields of the two union
+ operations must fully overlap.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.union_all(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.unionAll(right._j_table))
+
+ def intersect(self, right):
+ """
+ Intersects two :class:`Table`s with duplicate records removed. Intersect returns records
+ that exist in both tables. If a record is present in one or both tables more than once,
+ it is returned just once, i.e., the resulting table has no duplicate records. Similar to a
+ SQL INTERSECT. The fields of the two intersect operations must fully overlap.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.intersect(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.intersect(right._j_table))
+
+ def intersect_all(self, right):
+ """
+ Intersects two :class:`Table`s. IntersectAll returns records that exist in both tables.
+ If a record is present in both tables more than once, it is returned as many times as it
+ is present in both tables, i.e., the resulting table might have duplicate records. Similar
+ to an SQL INTERSECT ALL. The fields of the two intersect operations must fully overlap.
+
+ .. note::
+ Both tables must be bound to the same :class:`TableEnvironment`.
+
+ Example:
+ ::
+ >>> left.intersect_all(right)
+
+ :param right: Right table.
+ :return: Result table.
+ """
+ return Table(self._j_table.intersectAll(right._j_table))
+
+ def order_by(self, fields):
+ """
+ Sorts the given :class:`Table`. Similar to SQL ORDER BY.
+ The resulting Table is sorted globally sorted across all parallel partitions.
+
+ Example:
+ ::
+ >>> tab.order_by("name.desc")
+
+ :param fields: Order fields expression string,
+ :return: Result table.
+ """
+ return Table(self._j_table.orderBy(fields))
+
+ def offset(self, offset):
+ """
+ Limits a sorted result from an offset position.
+ Similar to a SQL OFFSET clause. Offset is technically part of the Order By operator and
+ thus must be preceded by it.
+ :func:`~pyflink.table.Table.offset` can be combined with a subsequent
+ :func:`~pyflink.table.Table.fetch` call to return n rows after skipping the first o rows.
+
+ Example:
+ ::
+ # skips the first 3 rows and returns all following rows.
+ >>> tab.order_by("name.desc").offset(3)
+ # skips the first 10 rows and returns the next 5 rows.
+ >>> tab.order_by("name.desc").offset(10).fetch(5)
+
+ :param offset: Number of records to skip.
+ :return: Result table.
+ """
+ return Table(self._j_table.offset(offset))
+
+ def fetch(self, fetch):
+ """
+ Limits a sorted result to the first n rows.
+ Similar to a SQL FETCH clause. Fetch is technically part of the Order By operator and
+ thus must be preceded by it.
+ :func:`~pyflink.table.Table.offset` can be combined with a preceding
+ :func:`~pyflink.table.Table.fetch` call to return n rows after skipping the first o rows.
+
+ Example:
+
+ Returns the first 3 records.
+ ::
+ >>> tab.order_by("name.desc").fetch(3)
+
+ Skips the first 10 rows and returns the next 5 rows.
+ ::
+ >>> tab.order_by("name.desc").offset(10).fetch(5)
+
+ :param fetch: The number of records to return. Fetch must be >= 0.
+ :return: Result table.
+ """
+ return Table(self._j_table.fetch(fetch))
+
+ def window(self, *windows):
+ """
Review comment:
I think `window()` and `over_window()` make more sense for Python, because it makes it more clear when you read the code. As @sunjincheng121 suggested.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services