You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/05/13 05:10:08 UTC
[flink] branch master updated: [FLINK-27271][python] Introduce Globalwindows window allocator
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ce6ee9cee34 [FLINK-27271][python] Introduce Globalwindows window allocator
ce6ee9cee34 is described below
commit ce6ee9cee34ff2b51acbd2928de943c6d2b26024
Author: zhangjingcun <cu...@163.com>
AuthorDate: Wed May 11 10:22:32 2022 +0800
[FLINK-27271][python] Introduce Globalwindows window allocator
This closes #19696.
---
.../docs/dev/datastream/operators/windows.md | 10 ++
.../docs/dev/datastream/operators/windows.md | 10 ++
flink-python/pyflink/datastream/data_stream.py | 7 +-
.../pyflink/datastream/tests/test_window.py | 33 +++++-
flink-python/pyflink/datastream/window.py | 124 ++++++++++++++++++++-
.../pyflink/fn_execution/coder_impl_fast.pyx | 14 ++-
.../pyflink/fn_execution/coder_impl_slow.py | 13 +++
flink-python/pyflink/fn_execution/coders.py | 9 ++
8 files changed, 216 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/dev/datastream/operators/windows.md b/docs/content.zh/docs/dev/datastream/operators/windows.md
index 0f22b6d0e27..2f9fa9393ac 100644
--- a/docs/content.zh/docs/dev/datastream/operators/windows.md
+++ b/docs/content.zh/docs/dev/datastream/operators/windows.md
@@ -470,6 +470,16 @@ input
.<windowed transformation>(<window function>)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+
+input \
+ .key_by(<key selector>) \
+ .window(GlobalWindows.create()) \
+ .<windowed transformation>(<window function>)
+```
+{{< /tab >}}
{{< /tabs >}}
## 窗口函数（Window Functions）
diff --git a/docs/content/docs/dev/datastream/operators/windows.md b/docs/content/docs/dev/datastream/operators/windows.md
index 19cdc4426b5..a4b6a492d49 100644
--- a/docs/content/docs/dev/datastream/operators/windows.md
+++ b/docs/content/docs/dev/datastream/operators/windows.md
@@ -490,6 +490,16 @@ input
.<windowed transformation>(<window function>)
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+input = ... # type: DataStream
+
+input \
+ .key_by(<key selector>) \
+ .window(GlobalWindows.create()) \
+ .<windowed transformation>(<window function>)
+```
+{{< /tab >}}
{{< /tabs >}}
## Window Functions
diff --git a/flink-python/pyflink/datastream/data_stream.py b/flink-python/pyflink/datastream/data_stream.py
index 73a0e169998..5f82dee0014 100644
--- a/flink-python/pyflink/datastream/data_stream.py
+++ b/flink-python/pyflink/datastream/data_stream.py
@@ -43,7 +43,8 @@ from pyflink.datastream.state import ValueStateDescriptor, ValueState, ListState
from pyflink.datastream.utils import convert_to_python_obj
from pyflink.datastream.window import (CountTumblingWindowAssigner, CountSlidingWindowAssigner,
CountWindowSerializer, TimeWindowSerializer, Trigger,
- WindowAssigner, WindowOperationDescriptor)
+ WindowAssigner, WindowOperationDescriptor,
+ GlobalWindowSerializer)
from pyflink.java_gateway import get_gateway
__all__ = ['CloseableIterator', 'DataStream', 'KeyedStream', 'ConnectedStreams', 'WindowedStream',
@@ -2138,6 +2139,10 @@ def _get_one_input_stream_operator(data_stream: DataStream,
elif isinstance(window_serializer, CountWindowSerializer):
j_namespace_serializer = \
gateway.jvm.org.apache.flink.table.runtime.operators.window.CountWindow.Serializer()
+ elif isinstance(window_serializer, GlobalWindowSerializer):
+ j_namespace_serializer = \
+ gateway.jvm.org.apache.flink.streaming.api.windowing.windows.GlobalWindow \
+ .Serializer()
else:
j_namespace_serializer = \
gateway.jvm.org.apache.flink.streaming.api.utils.ByteArrayWrapperSerializer()
diff --git a/flink-python/pyflink/datastream/tests/test_window.py b/flink-python/pyflink/datastream/tests/test_window.py
index 9e6004d45a5..e91e6482b65 100644
--- a/flink-python/pyflink/datastream/tests/test_window.py
+++ b/flink-python/pyflink/datastream/tests/test_window.py
@@ -27,7 +27,8 @@ from pyflink.datastream.output_tag import OutputTag
from pyflink.datastream.window import (TumblingEventTimeWindows,
SlidingEventTimeWindows, EventTimeSessionWindows,
CountSlidingWindowAssigner, SessionWindowTimeGapExtractor,
- CountWindow, PurgingTrigger, EventTimeTrigger, TimeWindow)
+ CountWindow, PurgingTrigger, EventTimeTrigger, TimeWindow,
+ GlobalWindows, CountTrigger)
from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction
from pyflink.java_gateway import get_gateway
from pyflink.testing.test_case_utils import PyFlinkStreamingTestCase
@@ -410,6 +411,36 @@ class WindowTests(PyFlinkStreamingTestCase):
side_expected = ['+I[a, 4]']
self.assert_equals_sorted(side_expected, side_sink.get_results())
+ def test_global_window_with_purging_trigger(self):
+ self.env.set_parallelism(1)
+ data_stream = self.env.from_collection([
+ ('hi', 1), ('hi', 1), ('hi', 1), ('hi', 1), ('hi', 1), ('hi', 1), ('hi', 1)],
+ type_info=Types.TUPLE([Types.STRING(), Types.INT()])) # type: DataStream
+
+ watermark_strategy = WatermarkStrategy.for_monotonous_timestamps() \
+ .with_timestamp_assigner(SecondColumnTimestampAssigner())
+
+ class MyProcessFunction(ProcessWindowFunction):
+
+ def clear(self, context: ProcessWindowFunction.Context) -> None:
+ pass
+
+ def process(self, key, context: ProcessWindowFunction.Context,
+ elements: Iterable[Tuple[str, int]]) -> Iterable[tuple]:
+ return [(key, len([e for e in elements]))]
+
+ data_stream.assign_timestamps_and_watermarks(watermark_strategy) \
+ .key_by(lambda x: x[0], key_type=Types.STRING()) \
+ .window(GlobalWindows.create()) \
+ .trigger(PurgingTrigger.of(CountTrigger.of(2))) \
+ .process(MyProcessFunction(), Types.TUPLE([Types.STRING(), Types.INT()])) \
+ .add_sink(self.test_sink)
+
+ self.env.execute('test_global_window_with_purging_trigger')
+ results = self.test_sink.get_results()
+ expected = ['(hi,2)', '(hi,2)', '(hi,2)']
+ self.assert_equals_sorted(expected, results)
+
class SecondColumnTimestampAssigner(TimestampAssigner):
diff --git a/flink-python/pyflink/datastream/window.py b/flink-python/pyflink/datastream/window.py
index 9da5cab96e0..afbbc99664d 100644
--- a/flink-python/pyflink/datastream/window.py
+++ b/flink-python/pyflink/datastream/window.py
@@ -33,6 +33,7 @@ from pyflink.metrics import MetricGroup
__all__ = ['Window',
'TimeWindow',
'CountWindow',
+ 'GlobalWindow',
'TumblingProcessingTimeWindows',
'TumblingEventTimeWindows',
'SlidingProcessingTimeWindows',
@@ -51,11 +52,13 @@ __all__ = ['Window',
'ProcessingTimeTrigger',
'ContinuousEventTimeTrigger',
'ContinuousProcessingTimeTrigger',
+ 'NeverTrigger',
'PurgingTrigger',
'CountTrigger',
'TimeWindowSerializer',
'CountWindowSerializer',
- 'SessionWindowTimeGapExtractor']
+ 'SessionWindowTimeGapExtractor',
+ 'GlobalWindows']
def long_to_int_with_bit_mixing(x: int) -> int:
@@ -197,6 +200,30 @@ class CountWindow(Window):
return "CountWindow(id={})".format(self.id)
+class GlobalWindow(Window):
+ """
+ The default window into which all data is placed GlobalWindows.
+ """
+ def __init__(self):
+ super(GlobalWindow, self).__init__()
+
+ @staticmethod
+ def get() -> 'GlobalWindow':
+ return GlobalWindow()
+
+ def max_timestamp(self) -> int:
+ return MAX_LONG_VALUE
+
+ def __eq__(self, other):
+ return self.__class__ == other.__class__
+
+ def __hash__(self):
+ return 0
+
+ def __repr__(self):
+ return "GlobalWindow"
+
+
class TimeWindowSerializer(TypeSerializer[TimeWindow]):
"""
The serializer used to write the TimeWindow type.
@@ -244,6 +271,31 @@ class CountWindowSerializer(TypeSerializer[CountWindow]):
return coders.CountWindowCoder()
+class GlobalWindowSerializer(TypeSerializer[GlobalWindow]):
+ """
+ A TypeSerializer for GlobalWindow.
+ """
+
+ def __init__(self):
+ self._underlying_coder = None
+
+ def serialize(self, element: GlobalWindow, stream: BytesIO) -> None:
+ if self._underlying_coder is None:
+ self._underlying_coder = self._get_coder().get_impl()
+ bytes_data = self._underlying_coder.encode(element)
+ stream.write(bytes_data)
+
+ def deserialize(self, stream: BytesIO) -> GlobalWindow:
+ if self._underlying_coder is None:
+ self._underlying_coder = self._get_coder().get_impl()
+ bytes_data = stream.read(8)
+ return self._underlying_coder.decode(bytes_data)
+
+ def _get_coder(self):
+ from pyflink.fn_execution import coders
+ return coders.GlobalWindowCoder()
+
+
T = TypeVar('T')
W = TypeVar('W')
W2 = TypeVar('W2')
@@ -965,6 +1017,41 @@ class CountTrigger(Trigger[T, CountWindow]):
count_state.clear()
+class NeverTrigger(Trigger[T, GlobalWindow]):
+ """
+ A trigger that never fires, as default Trigger for GlobalWindows.
+ """
+
+ def on_element(self,
+ element: T,
+ timestamp: int,
+ window: GlobalWindow,
+ ctx: 'Trigger.TriggerContext') -> TriggerResult:
+ return TriggerResult.CONTINUE
+
+ def on_processing_time(self,
+ time: int,
+ window: GlobalWindow,
+ ctx: 'Trigger.TriggerContext') -> TriggerResult:
+ return TriggerResult.CONTINUE
+
+ def on_event_time(self,
+ time: int,
+ window: GlobalWindow,
+ ctx: 'Trigger.TriggerContext') -> TriggerResult:
+ return TriggerResult.CONTINUE
+
+ def on_merge(self,
+ window: GlobalWindow,
+ ctx: 'Trigger.OnMergeContext') -> None:
+ pass
+
+ def clear(self,
+ window: GlobalWindow,
+ ctx: 'Trigger.TriggerContext') -> None:
+ pass
+
+
class CountTumblingWindowAssigner(WindowAssigner[T, CountWindow]):
"""
A WindowAssigner that windows elements into fixed-size windows based on the count number
@@ -1589,3 +1676,38 @@ class DynamicEventTimeSessionWindows(MergingWindowAssigner[T, TimeWindow]):
def __repr__(self):
return "DynamicEventTimeSessionWindows(%s)" % self._session_gap
+
+
+class GlobalWindows(WindowAssigner[T, GlobalWindow]):
+ """
+ A WindowAssigner that assigns all elements to the same GlobalWindow.
+ """
+
+ def __init__(self) -> None:
+ super().__init__()
+
+ def assign_windows(self,
+ element: T,
+ timestamp: int,
+ context: 'WindowAssigner.WindowAssignerContext') -> Collection[GlobalWindow]:
+ return [GlobalWindow.get()]
+
+ @staticmethod
+ def create() -> 'GlobalWindows':
+ """
+ Creates a new GlobalWindows WindowAssigner that assigns all elements to the
+ same GlobalWindow.
+ """
+ return GlobalWindows()
+
+ def get_default_trigger(self, env) -> Trigger[T, GlobalWindow]:
+ return NeverTrigger()
+
+ def get_window_serializer(self) -> TypeSerializer[GlobalWindow]:
+ return GlobalWindowSerializer()
+
+ def is_event_time(self) -> bool:
+ return False
+
+ def __repr__(self) -> str:
+ return "GlobalWindows()"
diff --git a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
index 518fcf826c5..4d3dd19ea19 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
+++ b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
@@ -31,7 +31,7 @@ import cloudpickle
from pyflink.common import Row, RowKind
from pyflink.common.time import Instant
-from pyflink.datastream.window import CountWindow, TimeWindow
+from pyflink.datastream.window import CountWindow, TimeWindow, GlobalWindow
from pyflink.fn_execution.ResettableIO import ResettableIO
from pyflink.table.utils import pandas_to_arrow, arrow_to_pandas
@@ -909,6 +909,18 @@ cdef class CountWindowCoderImpl(FieldCoderImpl):
cpdef decode_from_stream(self, InputStream in_stream, size_t size):
return CountWindow(in_stream.read_int64())
+cdef class GlobalWindowCoderImpl(FieldCoderImpl):
+ """
+ A coder for GlobalWindow.
+ """
+
+ cpdef encode_to_stream(self, value, OutputStream out_stream):
+ out_stream.write_byte(0)
+
+ cpdef decode_from_stream(self, InputStream in_stream, size_t size):
+ in_stream.read_byte()
+ return GlobalWindow()
+
cdef class DataViewFilterCoderImpl(FieldCoderImpl):
"""
A coder for CountWindow.
diff --git a/flink-python/pyflink/fn_execution/coder_impl_slow.py b/flink-python/pyflink/fn_execution/coder_impl_slow.py
index 75fd258d94f..9a847c417f6 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_slow.py
+++ b/flink-python/pyflink/fn_execution/coder_impl_slow.py
@@ -800,6 +800,19 @@ class CountWindowCoderImpl(FieldCoderImpl):
return CountWindow(in_stream.read_int64())
+class GlobalWindowCoderImpl(FieldCoderImpl):
+ """
+ A coder for CountWindow.
+ """
+
+ def encode_to_stream(self, value, out_stream: OutputStream):
+ out_stream.write_byte(0)
+
+ def decode_from_stream(self, in_stream: InputStream, length=0):
+ in_stream.read_byte()
+ return GlobalWindowCoderImpl()
+
+
class DataViewFilterCoderImpl(FieldCoderImpl):
"""
A coder for data view filter.
diff --git a/flink-python/pyflink/fn_execution/coders.py b/flink-python/pyflink/fn_execution/coders.py
index 7e43f78b2b2..fd094d77f46 100644
--- a/flink-python/pyflink/fn_execution/coders.py
+++ b/flink-python/pyflink/fn_execution/coders.py
@@ -585,6 +585,15 @@ class CountWindowCoder(FieldCoder):
return coder_impl.CountWindowCoderImpl()
+class GlobalWindowCoder(FieldCoder):
+ """
+ Coder for GlobalWindow.
+ """
+
+ def get_impl(self):
+ return coder_impl.GlobalWindowCoderImpl()
+
+
class DataViewFilterCoder(FieldCoder):
"""
Coder for data view filter.