Timm0 commented on code in PR #29070:
URL: https://github.com/apache/flink/pull/29070#discussion_r3977167664
##########
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:
Changed the assertion to fix the potential instability
--
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]
