From cb999e9e80c32c32dc073293406970e9af7b9bd5 Mon Sep 17 00:00:00 2001 From: Don Wong Date: Wed, 13 Oct 2021 10:22:27 +0800 Subject: [PATCH 1/3] robinhood issues 514 --- faust/tables/base.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/faust/tables/base.py b/faust/tables/base.py index 91ba34939..6db7bb56b 100644 --- a/faust/tables/base.py +++ b/faust/tables/base.py @@ -366,16 +366,17 @@ async def _del_old_keys(self) -> None: assert window for partition, timestamps in self._partition_timestamps.items(): while timestamps and window.stale( - timestamps[0], self._partition_latest_timestamp[partition] - ): + timestamps[0], + self._partition_latest_timestamp[partition]): timestamp = heappop(timestamps) - keys_to_remove = self._partition_timestamp_keys.pop( - (partition, timestamp), None - ) + keysList = [self._partition_timestamp_keys.get((partition, window_range[1])) for window_range in self._window_ranges(timestamp)] + keys_to_remove = self._partition_timestamp_keys.pop((partition, timestamp), None) if keys_to_remove: + windowData = [item for keys in keysList for key in keys for item in self.data.get(key, None)] for key in keys_to_remove: value = self.data.pop(key, None) - await self.on_window_close(key, value) + if key[1][0] > self.last_closed_window: + await self._on_window_close(key, windowData) self.last_closed_window = max( self.last_closed_window, max(key[1][0] for key in keys_to_remove), From 086814387535f7c4dc9f465d3d64d4f77f809f25 Mon Sep 17 00:00:00 2001 From: Don Wong Date: Wed, 13 Oct 2021 10:25:44 +0800 Subject: [PATCH 2/3] add hopping example --- examples/windowing/hopping_2.py | 103 ++++++++++++++++++++++++++++++++ 1 file changed, 103 insertions(+) create mode 100644 examples/windowing/hopping_2.py diff --git a/examples/windowing/hopping_2.py b/examples/windowing/hopping_2.py new file mode 100644 index 000000000..39a0871fc --- /dev/null +++ b/examples/windowing/hopping_2.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python + +# In this exapmple we have a function `publish_every_2secs` publishing a +# message every 2 seconds to topic `hopping_topic`. +# We have created an agent `print_windowed_events` consuming events from +# `hopping_topic` that mutates the windowed table `hopping_table`. + +# `hopping_table` is a table with hopping (overlaping) windows. Each of +# its windows is 10 seconds of duration, and we create a new window every 5 +# seconds. +# |----------| +# |-----------| +# |-----------| +# |-----------| +# Since we produce an event every 2 seconds and our windows are 10 +# seconds of duration we expect different the following results per method +# called in `WindowWrapper`: +# - now(): Gets the closest value to current local time. It will always be +# between 1 and 3. +# - current(): Gets the value relative to the event's timestamp. It will +# always be between 1 and 3. +# - value(): Gets the value relative to default relative option. It will +# always be between 1 and 3. +# - delta(30): Gets the value of window 30 secs before the current event. For +# the first 20 seconds it will be 0 and after second 30 it will always be 5. + +import sys +from datetime import datetime, timedelta +from time import time +import faust + + +class RawModel(faust.Record): + date: datetime + value: float + + +class WinModel(faust.Record): + win: list + + +TOPIC = 'raw-event' +TABLE = 'hopping_table' +KAFKA = 'kafka://localhost:9092' +CLEANUP_INTERVAL = 10 +WINDOW = 6 +STEP = 3 +WINDOW_EXPIRES = 60 +PARTITIONS = 1 + +app = faust.App('windowed-hopping', broker=KAFKA, topic_partitions=PARTITIONS) + +app.conf.table_cleanup_interval = CLEANUP_INTERVAL +source = app.topic(TOPIC, value_type=RawModel) + + +def window_processor(key, events): + timestamp = key[1][0] + count = len(events) + + print( + f'processing window:' + f'{count} events,' + f'timestamp {timestamp}', + ) + + +hopping_table = ( + app.Table( + TABLE, + default=list, + partitions=PARTITIONS, + on_window_close=window_processor, + ) + .hopping(WINDOW, STEP, expires=timedelta(seconds=WINDOW_EXPIRES)) + .relative_to_field(RawModel.date) +) + + +@app.agent(source) +async def print_windowed_events(stream): + async for event in stream: + value_list = hopping_table['events'].value() + + if len(value_list) > 0: + event.value = value_list[-1].value + 1 + print("Receive message : " + str(event)) + + value_list.append(event) + hopping_table['events'] = value_list + + +@app.timer(0.1) +async def produce(): + value = 1 + await source.send(value=RawModel(value=value, date=int(time()))) + # print(f'Produce Message :: send messge {value}') + + +if __name__ == '__main__': + if len(sys.argv) < 2: + sys.argv.extend(['worker', '-l', 'info']) + app.main() From 4259492027827e7cc51f891e019170d816bd76f6 Mon Sep 17 00:00:00 2001 From: Don Wong Date: Wed, 13 Oct 2021 21:02:12 +0800 Subject: [PATCH 3/3] fix: 'Nonetype' object is not iterable problem --- faust/tables/base.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/faust/tables/base.py b/faust/tables/base.py index 6db7bb56b..0ce186a42 100644 --- a/faust/tables/base.py +++ b/faust/tables/base.py @@ -372,11 +372,11 @@ async def _del_old_keys(self) -> None: keysList = [self._partition_timestamp_keys.get((partition, window_range[1])) for window_range in self._window_ranges(timestamp)] keys_to_remove = self._partition_timestamp_keys.pop((partition, timestamp), None) if keys_to_remove: - windowData = [item for keys in keysList for key in keys for item in self.data.get(key, None)] + windowData = [item for keys in keysList if keys for key in keys for item in self.data.get(key, None)] for key in keys_to_remove: value = self.data.pop(key, None) if key[1][0] > self.last_closed_window: - await self._on_window_close(key, windowData) + await self.on_window_close(key, windowData) self.last_closed_window = max( self.last_closed_window, max(key[1][0] for key in keys_to_remove),