You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by hx...@apache.org on 2021/03/02 08:44:55 UTC
[flink] branch master updated: [FLINK-21202][python] Introduce
TimeWindow and CountWindow in PyFlink
This is an automated email from the ASF dual-hosted git repository.
hxb 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 1375fba [FLINK-21202][python] Introduce TimeWindow and CountWindow in PyFlink
1375fba is described below
commit 1375fba82b8ca7f34c7367882d78f6684ff89b6e
Author: huangxingbo <hx...@gmail.com>
AuthorDate: Sat Jan 30 13:33:02 2021 +0800
[FLINK-21202][python] Introduce TimeWindow and CountWindow in PyFlink
This closes #14816
---
.../fn_execution/beam/beam_coder_impl_slow.py | 23 ++++
.../pyflink/fn_execution/coder_impl_fast.pxd | 13 +++
.../pyflink/fn_execution/coder_impl_fast.pyx | 37 ++++++
flink-python/pyflink/fn_execution/coders.py | 19 ++-
.../pyflink/fn_execution/tests/test_coders.py | 10 +-
.../pyflink/fn_execution/tests/test_fast_coders.py | 13 +++
flink-python/pyflink/fn_execution/window.py | 128 +++++++++++++++++++++
7 files changed, 241 insertions(+), 2 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py b/flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py
index cb3570a..7f353e3 100644
--- a/flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py
+++ b/flink-python/pyflink/fn_execution/beam/beam_coder_impl_slow.py
@@ -28,11 +28,34 @@ from apache_beam.coders.coder_impl import StreamCoderImpl, create_InputStream, c
from pyflink.fn_execution.ResettableIO import ResettableIO
from pyflink.common import Row, RowKind
+from pyflink.fn_execution.window import TimeWindow, CountWindow
from pyflink.table.utils import pandas_to_arrow, arrow_to_pandas
ROW_KIND_BIT_SIZE = 2
+class TimeWindowCoderImpl(StreamCoderImpl):
+
+ def encode_to_stream(self, value: TimeWindow, stream, nested):
+ stream.write_bigendian_int64(value.start)
+ stream.write_bigendian_int64(value.end)
+
+ def decode_from_stream(self, stream, nested):
+ start = stream.read_bigendian_int64()
+ end = stream.read_bigendian_int64()
+ return TimeWindow(start, end)
+
+
+class CountWindowCoderImpl(StreamCoderImpl):
+
+ def encode_to_stream(self, value: CountWindow, stream, nested):
+ stream.write_bigendian_int64(value.id)
+
+ def decode_from_stream(self, stream, nested):
+ id = stream.read_bigendian_int64()
+ return CountWindow(id)
+
+
class FlattenRowCoderImpl(StreamCoderImpl):
def __init__(self, field_coders):
diff --git a/flink-python/pyflink/fn_execution/coder_impl_fast.pxd b/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
index 50ba5e5..4ecad43 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
+++ b/flink-python/pyflink/fn_execution/coder_impl_fast.pxd
@@ -127,6 +127,19 @@ cdef class DataStreamFlatMapCoderImpl(BaseCoderImpl):
cdef class DataStreamCoFlatMapCoderImpl(BaseCoderImpl):
cdef readonly object _single_field_coder
+cdef class WindowCoderImpl(BaseCoderImpl):
+ cdef size_t _tmp_output_pos
+ cdef char*_tmp_output_data
+
+ cpdef bytes encode_nested(self, value)
+ cdef void _encode_bigint(self, libc.stdint.int64_t v)
+
+cdef class TimeWindowCoderImpl(WindowCoderImpl):
+ pass
+
+cdef class CountWindowCoderImpl(WindowCoderImpl):
+ pass
+
cdef enum CoderType:
UNDEFINED = -1
SIMPLE = 0
diff --git a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
index 397b789..0d2a8b1 100644
--- a/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
+++ b/flink-python/pyflink/fn_execution/coder_impl_fast.pyx
@@ -25,6 +25,8 @@ from libc.string cimport memcpy
import datetime
import decimal
+
+from pyflink.fn_execution.window import TimeWindow, CountWindow
from pyflink.table import Row
from pyflink.table.types import RowKind
@@ -811,6 +813,41 @@ cdef class FlattenRowCoderImpl(BaseCoderImpl):
if self._field_coder_type:
free(self._field_coder_type)
+cdef class WindowCoderImpl(BaseCoderImpl):
+ def __init__(self):
+ self._tmp_output_pos = 0
+ self._tmp_output_data = <char*> malloc(16)
+
+ cpdef bytes encode_nested(self, value):
+ pass
+
+ cdef void _encode_bigint(self, libc.stdint.int64_t v):
+ self._tmp_output_data[self._tmp_output_pos] = <unsigned char> (v >> 56)
+ self._tmp_output_data[self._tmp_output_pos + 1] = <unsigned char> (v >> 48)
+ self._tmp_output_data[self._tmp_output_pos + 2] = <unsigned char> (v >> 40)
+ self._tmp_output_data[self._tmp_output_pos + 3] = <unsigned char> (v >> 32)
+ self._tmp_output_data[self._tmp_output_pos + 4] = <unsigned char> (v >> 24)
+ self._tmp_output_data[self._tmp_output_pos + 5] = <unsigned char> (v >> 16)
+ self._tmp_output_data[self._tmp_output_pos + 6] = <unsigned char> (v >> 8)
+ self._tmp_output_data[self._tmp_output_pos + 7] = <unsigned char> v
+ self._tmp_output_pos += 8
+
+ def __dealloc__(self):
+ if self._tmp_output_data:
+ free(self._tmp_output_data)
+
+cdef class TimeWindowCoderImpl(WindowCoderImpl):
+ cpdef bytes encode_nested(self, value: TimeWindow):
+ self._encode_bigint(value.start)
+ self._encode_bigint(value.end)
+ return self._tmp_output_data[:16]
+
+
+cdef class CountWindowCoderImpl(WindowCoderImpl):
+ cpdef bytes encode_nested(self, value: CountWindow):
+ self._encode_bigint(value.id)
+ return self._tmp_output_data[:8]
+
cdef class FieldCoder:
cpdef CoderType coder_type(self):
return UNDEFINED
diff --git a/flink-python/pyflink/fn_execution/coders.py b/flink-python/pyflink/fn_execution/coders.py
index efb25c9..6777037 100644
--- a/flink-python/pyflink/fn_execution/coders.py
+++ b/flink-python/pyflink/fn_execution/coders.py
@@ -32,7 +32,8 @@ except:
__all__ = ['RowCoder', 'BigIntCoder', 'TinyIntCoder', 'BooleanCoder',
'SmallIntCoder', 'IntCoder', 'FloatCoder', 'DoubleCoder',
'BinaryCoder', 'CharCoder', 'DateCoder', 'TimeCoder',
- 'TimestampCoder', 'BasicArrayCoder', 'PrimitiveArrayCoder', 'MapCoder', 'DecimalCoder']
+ 'TimestampCoder', 'BasicArrayCoder', 'PrimitiveArrayCoder', 'MapCoder', 'DecimalCoder',
+ 'TimeWindowCoder', 'CountWindowCoder']
# table coders
FLINK_SCALAR_FUNCTION_SCHEMA_CODER_URN = "flink:coder:schema:scalar_function:v1"
@@ -243,6 +244,22 @@ class DataStreamCoFlatMapCoder(BaseCoder):
return hash(self._field_coders)
+class TimeWindowCoder(BaseCoder):
+ """
+ Coder for TimeWindow.
+ """
+ def get_impl(self):
+ return coder_impl.TimeWindowCoderImpl()
+
+
+class CountWindowCoder(BaseCoder):
+ """
+ Coder for CountWindow.
+ """
+ def get_impl(self):
+ return coder_impl.CountWindowCoderImpl()
+
+
class FieldCoder(ABC):
def get_impl(self):
diff --git a/flink-python/pyflink/fn_execution/tests/test_coders.py b/flink-python/pyflink/fn_execution/tests/test_coders.py
index 6b0f9b0..96a3613 100644
--- a/flink-python/pyflink/fn_execution/tests/test_coders.py
+++ b/flink-python/pyflink/fn_execution/tests/test_coders.py
@@ -24,7 +24,9 @@ import unittest
from pyflink.fn_execution.coders import BigIntCoder, TinyIntCoder, BooleanCoder, \
SmallIntCoder, IntCoder, FloatCoder, DoubleCoder, BinaryCoder, CharCoder, DateCoder, \
TimeCoder, TimestampCoder, BasicArrayCoder, MapCoder, DecimalCoder, FlattenRowCoder, RowCoder, \
- LocalZonedTimestampCoder, BigDecimalCoder, TupleCoder, PrimitiveArrayCoder
+ LocalZonedTimestampCoder, BigDecimalCoder, TupleCoder, PrimitiveArrayCoder, TimeWindowCoder, \
+ CountWindowCoder
+from pyflink.fn_execution.window import TimeWindow, CountWindow
from pyflink.testing.test_case_utils import PyFlinkTestCase
try:
@@ -180,6 +182,12 @@ class CodersTest(PyFlinkTestCase):
data = (1, "Hello", "Hi")
self.check_coder(tuple_coder, data)
+ def test_window_coder(self):
+ coder = TimeWindowCoder()
+ self.check_coder(coder, TimeWindow(100, 1000))
+ coder = CountWindowCoder()
+ self.check_coder(coder, CountWindow(100))
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
diff --git a/flink-python/pyflink/fn_execution/tests/test_fast_coders.py b/flink-python/pyflink/fn_execution/tests/test_fast_coders.py
index 884a528..f3f7e82 100644
--- a/flink-python/pyflink/fn_execution/tests/test_fast_coders.py
+++ b/flink-python/pyflink/fn_execution/tests/test_fast_coders.py
@@ -20,6 +20,7 @@
import logging
import unittest
+from pyflink.fn_execution.window import TimeWindow, CountWindow
from pyflink.testing.test_case_utils import PyFlinkTestCase
try:
@@ -230,6 +231,18 @@ class CodersTest(PyFlinkTestCase):
row.set_row_kind(RowKind.DELETE)
self.check_cython_coder(python_field_coders, cython_field_coders, data)
+ def test_cython_time_window_coder(self):
+ fast_coder = coder_impl_fast.TimeWindowCoderImpl()
+ slow_coder = coder_impl.TimeWindowCoderImpl()
+ window = TimeWindow(100, 200)
+ self.assertEqual(fast_coder.encode_nested(window), slow_coder.encode_nested(window))
+
+ def test_cython_count_window_coder(self):
+ fast_coder = coder_impl_fast.CountWindowCoderImpl()
+ slow_coder = coder_impl.CountWindowCoderImpl()
+ window = CountWindow(100)
+ self.assertEqual(fast_coder.encode_nested(window), slow_coder.encode_nested(window))
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
diff --git a/flink-python/pyflink/fn_execution/window.py b/flink-python/pyflink/fn_execution/window.py
new file mode 100644
index 0000000..ce816ae
--- /dev/null
+++ b/flink-python/pyflink/fn_execution/window.py
@@ -0,0 +1,128 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import sys
+from abc import ABC, abstractmethod
+
+MAX_LONG_VALUE = sys.maxsize
+
+
+def long_to_int_with_bit_mixing(x: int) -> int:
+ x = (x ^ (x >> 30)) * 0xbf58476d1ce4e5b9
+ x = (x ^ (x >> 27)) * 0x94d049bb133111eb
+ x = x ^ (x >> 31)
+ return x
+
+
+def mod_inverse(x: int) -> int:
+ inverse = x * x * x
+ inverse *= 2 - x * inverse
+ inverse *= 2 - x * inverse
+ inverse *= 2 - x * inverse
+ return inverse
+
+
+class Window(ABC):
+ """
+ Window is a grouping of elements into finite buckets. Windows have a maximum timestamp
+ which means that, at some point, all elements that go into one window will have arrived.
+ """
+
+ @abstractmethod
+ def max_timestamp(self) -> int:
+ pass
+
+
+class TimeWindow(Window):
+ """
+ Window that represents a time interval from start (inclusive) to end (exclusive).
+ """
+
+ def __init__(self, start: int, end: int):
+ super(TimeWindow, self).__init__()
+ self.start = start
+ self.end = end
+
+ def max_timestamp(self) -> int:
+ return self.end - 1
+
+ def intersects(self, other: 'TimeWindow') -> bool:
+ """
+ Returns True if this window intersects the given window.
+ """
+ return self.start <= other.end and self.end >= other.start
+
+ def cover(self, other: 'TimeWindow') -> 'TimeWindow':
+ """
+ Returns the minimal window covers both this window and the given window.
+ """
+ return TimeWindow(min(self.start, other.start), max(self.end, other.end))
+
+ @staticmethod
+ def get_window_start_with_offset(timestamp: int, offset: int, window_size: int):
+ """
+ Method to get the window start for a timestamp.
+
+ :param timestamp: epoch millisecond to get the window start.
+ :param offset: The offset which window start would be shifted by.
+ :param window_size: The size of the generated windows.
+ :return: window start
+ """
+ return timestamp - (timestamp - offset + window_size) % window_size
+
+ def __hash__(self):
+ return self.start + mod_inverse((self.end << 1) + 1)
+
+ def __eq__(self, other):
+ return self.__class__ == other.__class__ and self.end == other.end \
+ and self.start == other.start
+
+ def __lt__(self, other: 'TimeWindow'):
+ if not isinstance(other, TimeWindow):
+ raise Exception("Does not support comparison with non-TimeWindow %s" % other)
+
+ return self.start == other.start and self.end < other.end or self.start < other.start
+
+ def __le__(self, other: 'TimeWindow'):
+ return self.__eq__(other) and self.__lt__(other)
+
+ def __repr__(self):
+ return "TimeWindow(start={}, end={})".format(self.start, self.end)
+
+
+class CountWindow(Window):
+ """
+ A Window that represents a count window. For each count window, we will assign a unique
+ id. Thus this CountWindow can act as namespace part in state. We can attach data to each
+ different CountWindow.
+ """
+
+ def __init__(self, id: int):
+ super(CountWindow, self).__init__()
+ self.id = id
+
+ def max_timestamp(self) -> int:
+ return MAX_LONG_VALUE
+
+ def __hash__(self):
+ return long_to_int_with_bit_mixing(self.id)
+
+ def __eq__(self, other):
+ return self.__class__ == other.__class__ and self.id == other.id
+
+ def __repr__(self):
+ return "CountWindow(id={})".format(self.id)