--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/python/examples/wait_click_join.md - https://ytsaurus.tech/docs/ru/flow/python/examples/wait_click_join.md - href: https://ytsaurus.tech/docs/ru/flow/python/examples/wait_click_join.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Wait Click Join в YTsaurus Flow (Python) Пример join-[пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline), объединяющего [потоки](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) показов (hit) и действий (action) по ключу с использованием [таймеров](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timer) и external [state](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state). Python-реализация аналогичного [C++ примера](https://ytsaurus.tech/docs/ru/flow/cpp/examples/wait_click_join.md). [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/wait_click_join) ## Структура Пайплайн состоит из одного transform-компьютейшена `join`, который: 1. Получает сообщения из двух входных стримов: `hit` и `action`. 2. Накапливает данные во внешнем стейте. 3. Устанавливает таймер на окончание окна ожидания. 4. При срабатывании таймера генерирует join-результат или очищает стейт. ## `__main__.py` ```python def main(): log.info("Starting companion execution") pipeline = Pipeline() pipeline.add("join", JoinProcessFunction()) log.info("Starting pipeline...") pipeline.run() log.info("Pipeline completed") ``` ## `join_process_function.py` Основная логика join-а. Функция обрабатывает два стрима и использует таймеры для оконной агрегации: ### `on_message` ```python def on_message(self, message, output, ctx): stream_id = message.stream_id if stream_id == "hit": hit_id = message.payload["hit_id"] hit_time = message.payload["hit_time"] hit_payload = message.payload["hit_payload"] elif stream_id == "action": hit_id = message.payload["hit_id"] hit_time = message.payload["hit_time"] else: raise ValueError(f"Unknown streamId: {stream_id}") log.debug("Process message (Key: %s, HitId: %s, HitTime: %s)", message.key, hit_id, hit_time) wait_time = _parse_duration_seconds(ctx.parameters["wait_for_actions"]) max_time = hit_time + wait_time # All messages out of wait range would be dropped. if message.event_timestamp >= max_time: log.warning( "Message is out of wait range, skip (HitId: %s, HitTime: %s, MaxTime: %s)", hit_id, hit_time, max_time, ) return # Drop late data. if message.event_timestamp < ctx.min_watermark: log.warning( "Late message, skip (HitId: %s, EventTimestamp: %s, InputEventWatermark: %s)", hit_id, message.event_timestamp, ctx.min_watermark, ) return state = ctx.external_state("/join-state", message) builder = state.to_builder() if stream_id == "hit": log.debug("Process hit message (HitId: %s, HitPayload: %s)", hit_id, hit_payload) builder.set("hit_payload", hit_payload) else: action_time = message.payload["action_time"] is_click = message.payload["is_click"] log.debug("Process action message (HitId: %s, ActionTime: %s)", hit_id, action_time) if is_click: builder.set("click_time", action_time) else: builder.set("show_time", action_time) state.set(builder.finish()) log.debug("Add timer for Key (MaxTime: %s, HitTime: %s)", max_time, hit_time) output.add_timer(max_time, hit_time) ``` ### `on_timer` ```python def on_timer(self, timer, output, ctx): hit_id = timer.key["hit_id"] log.debug("Process timer (KeyHitId: %s, Timer: %s)", hit_id, timer) state = ctx.external_state("/join-state", timer) log.debug("Process state (KeyHitId: %s, State: %s)", hit_id, state) if timer.stream_id == "timer": show_time = state.get("show_time") hit_payload = state.get("hit_payload") if show_time is not None and hit_payload is not None and show_time != 0: log.debug("Click join: %s", state) hit_time = timer.key["hit_time"] click_time = state.get("click_time") builder = ctx.message_builder("joined_action") builder.set("hit_id", hit_id) builder.set("hit_time", hit_time) builder.set("show_time", show_time) builder.set("hit_payload", hit_payload) if click_time is not None and click_time != 0: builder.set("is_click", True) builder.set("click_time", click_time) else: builder.set("is_click", False) builder.set("click_time", 0) output.add_message(builder.finish()) state.clear() else: raise ValueError(f"Unknown streamId: {timer.stream_id}") ``` ## Ключевые паттерны - **External state** с `ctx.external_state("/join-state", message)`: паттерн `to_builder()` / `set()` / `clear()` для накопления данных из нескольких стримов. - **Таймеры** через `output.add_timer(max_time, hit_time)`: установка таймера с `trigger_timestamp` (когда сработает) и `event_timestamp` (привязка к событию). - **Фильтрация запоздавших данных**: `message.event_timestamp < ctx.min_watermark` для отброса late data (подробнее о [вотермарках](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timestamps-and-watermarks)). - **Параметры компьютейшена**: `ctx.parameters["wait_for_actions"]` для чтения конфигурации из спеки. - **MessageBuilder**: `ctx.message_builder("joined_action")` для создания выходных сообщений с заданной схемой. - **Обработка нескольких стримов**: ветвление по `message.stream_id` для различной логики обработки hit и action. ## См. также - [Быстрый старт (Python)](https://ytsaurus.tech/docs/ru/flow/python/getting-started.md) - [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md)