dianfu commented on code in PR #29070:
URL: https://github.com/apache/flink/pull/29070#discussion_r3966117356
##########
flink-python/pyflink/dataframe/tests/test_dataframe.py:
##########
@@ -1961,5 +2070,111 @@ def
test_grouped_aggregation_with_batch_table_environment(self):
)
+class DataFrameWindowITTests(PyFlinkStreamDataFrameTestCase):
+ @classmethod
+ def setUpClass(cls):
+ super().setUpClass()
+ cls.t_env.get_config().set("table.exec.resource.default-parallelism",
"1")
+
+ def _rowtime_source(self):
+ input_path = os.path.join(self.tempdir, "events.csv")
+ with open(input_path, "w", encoding="utf-8") as events:
+ events.write("1,10,0\n")
+ events.write("1,20,60000\n")
+ events.write("1,40,600000\n")
+ return pf.read_generic(
+ "filesystem",
+ schema={
+ "id": DataType.int64(),
+ "amount": DataType.int64(),
+ "ts_millis": DataType.int64(),
+ },
+ options={"path": input_path, "format": "csv"},
+ computed_columns={"event_time": "TO_TIMESTAMP_LTZ(ts_millis, 3)"},
+ watermark=("event_time", "event_time - INTERVAL '1' SECOND"),
+ )
+
+ def _proctime_source(self):
+ input_path = os.path.join(self.tempdir, "proctime_events.csv")
+ with open(input_path, "w", encoding="utf-8") as events:
+ events.write("1,10\n")
+ events.write("1,20\n")
+ events.write("1,40\n")
+ return pf.read_generic(
+ "filesystem",
+ schema={
+ "id": DataType.int64(),
+ "amount": DataType.int64(),
+ },
+ options={"path": input_path, "format": "csv"},
+ computed_columns={"proc_time": "PROCTIME()"},
+ )
+
+ def test_tumble_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .tumble(on="event_time", size=timedelta(minutes=10))
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
40])
+
+ def test_hop_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .hop(
+ on="event_time",
+ slide=timedelta(minutes=5),
+ size=timedelta(minutes=10),
+ )
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
30, 40, 40])
+
+ def test_cumulate_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .cumulate(
+ on="event_time",
+ step=timedelta(minutes=5),
+ size=timedelta(minutes=10),
+ )
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+
+ self.assertEqual(sorted(row[-1] for row in windowed.collect()), [30,
30, 40, 40])
+
+ def test_session_window_aggregation(self):
+ windowed = (
+ self._rowtime_source()
+ .session(on="event_time", gap=timedelta(minutes=5))
+ .group_by("window_start", "window_end", "id")
+ .agg(pf.col("amount").sum.alias("total"))
+ )
+ rows = self._materialize(windowed, key=["window_start", "window_end",
"id"])
+
+ self.assertEqual(sorted(row[-1] for row in rows), [30, 40])
+
+ def test_tumble_processing_time(self):
Review Comment:
This test may be unstable around a processing-time window boundary. For
example, if some records are processed at 10:09:59.999 and the remaining
records at 10:10:00.001, they will belong to different 10-minute windows and
this assertion will fail.
##########
flink-python/pyflink/dataframe/dataframe.py:
##########
@@ -560,6 +553,111 @@ def drop_duplicates(
distinct = drop_duplicates
unique = drop_duplicates
+ # ======================== Windowing ========================
+
+ @PublicEvolving()
+ def tumble(
+ self,
+ *,
+ on: Union[str, Expression],
+ size: Union["datetime.timedelta", Expression],
+ ) -> "DataFrame":
+ """
+ Assign rows to fixed-size, non-overlapping (tumbling) windows.
+
+ Appends ``window_start``, ``window_end`` and ``window_time`` and
returns an ordinary
+ DataFrame.
+
+ :param on: An existing event-time or processing-time column.
+ :param size: Window length.
+ :return: A new DataFrame with the window columns appended.
+ :raises TypeError: If ``on`` or ``size`` has an unsupported type.
Review Comment:
Could we add some examples for these API?
--
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]
