Я читаю события из PubSub, и цель состоит в том, чтобы сгруппировать их в окна. Я хотел бы, чтобы конец каждого окна совпадал с минутами 0, 15, 30 и 45 каждого часа.
Поскольку это потоковая работа, ее можно запустить в любое время, и я хотел бы найти способ выровнять размер первого окна со следующими.
Это будет поток:
- Запустить работу
- Определите как
window_sizeвремя, оставшееся между этим моментом и следующей четвертью часа. - Начиная с конца этого первого окна, установите
window_size = int(15*60)(секунды).
Например:
- Запустить работу
- Сейчас 11:18, так что исправь
window_size = (11:30-11:18).seconds - Когда это первое окно закончится, установите
window_size = int(15*60)(секунды)
В одном из примеров, предоставленных Google, конвейер, работающий с оконным управлением, определяется следующим образом, где window_size - это параметр, переданный в качестве ввода пользователем:
def expand(self, pcoll):
return (
pcoll
| "Window into Fixed Intervals" >> beam.WindowInto(window.FixedWindows(self.window_size))
| "Add Key" >> beam.Map(lambda elem: (None, elem))
| "Groupby" >> beam.GroupByKey()
| "Abandon Key" >> beam.MapTuple(lambda _, val: val)
)
