--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/go/examples/wait_click_join.md - https://ytsaurus.tech/docs/ru/flow/go/examples/wait_click_join.md - href: https://ytsaurus.tech/docs/ru/flow/go/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 (Go) Пример 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) и внешний [стейт](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state). Go-реализация аналогичного [примера на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/wait_click_join.md). [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/wait_click_join) ## Структура {#structure} Пайплайн состоит из трёх [компьютейшенов](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation): - `hit_reader` и `action_reader` — нативные сорсы (`TSwiftPassthroughOrderedSourceComputation`), объявленные прямо в [спеке](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec): они читают две очереди и публикуют стримы `hit` и `action`. Go-кода у них нет. Каждый сорс объявляет `watermark_strategy`, поэтому [вотермарк](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timestamps-and-watermarks) пайплайна считается по времени события. - `join` (`joinFunction`) — transform-компьютейшен, который обслуживает компаньон. Компьютейшен `join` работает так: 1. Получает сообщения из двух входных стримов: `hit` и `action`. 2. Накапливает окно одного показа во внешнем стейте `/join-state`. 3. Взводит таймер на момент закрытия окна ожидания. 4. По срабатыванию таймера публикует join-результат в стрим `joined_action` либо просто очищает стейт. Оба входных стрима группируются по `hit_id` и `hit_time`, поэтому всё, что относится к одному показу, попадает в один ключ. Длину окна задаёт параметр компьютейшена `wait_for_actions` из спеки. ## `main.go` {#main-go} Точка входа: создание пайплайна, регистрация единственного компьютейшена и запуск. [Исходный код: `main.go`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/wait_click_join/main.go) ```go type actionMessage struct { flow.YSONMessage HitID string `yson:"hit_id"` HitTime uint64 `yson:"hit_time"` IsClick bool `yson:"is_click"` ActionTime uint64 `yson:"action_time"` } type hitMessage struct { flow.YSONMessage HitID string `yson:"hit_id"` HitTime uint64 `yson:"hit_time"` HitPayload string `yson:"hit_payload"` } type joinedActionMessage struct { flow.YSONMessage HitID string `yson:"hit_id"` HitTime uint64 `yson:"hit_time"` IsClick bool `yson:"is_click"` ShowTime uint64 `yson:"show_time"` ClickTime uint64 `yson:"click_time"` HitPayload string `yson:"hit_payload"` } func main() { pipeline := flow.NewPipeline() pipeline.AddStreams( flow.NewYSONStream[actionMessage](actionStream), flow.NewYSONStream[hitMessage](hitStream), flow.NewYSONStream[joinedActionMessage](joinedStream), ) pipeline.Add(flow.NewRowComputation("join", &joinFunction{})) if err := pipeline.Run(); err != nil { fmt.Fprintf(os.Stderr, "wait_click_join: %v\n", err) os.Exit(1) } } ``` ## `join_function.go` {#join-function-go} Основная логика join-а: функция обрабатывает два стрима и использует таймеры для оконной агрегации. ### `OnMessage` {#on-message} Отбрасывает запоздавшие сообщения, дописывает своё поле в окно показа и взводит таймер на закрытие окна: ```go type joinFunction struct{} var ( _ flow.RowFunction = (*joinFunction)(nil) _ flow.RowTimerFunction = (*joinFunction)(nil) ) func (*joinFunction) OnMessage( ctx context.Context, rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector, ) error { var hit *hitMessage var action *actionMessage var hitTime uint64 switch msg.StreamID { case hitStream: hit = &hitMessage{} if err := msg.ConvertTo(hit); err != nil { return err } hitTime = hit.HitTime case actionStream: action = &actionMessage{} if err := msg.ConvertTo(action); err != nil { return err } hitTime = action.HitTime default: return fmt.Errorf("unhandled stream %q", msg.StreamID) } wait, err := waitForActions(rt) if err != nil { return err } closeTime := hitTime + wait if msg.EventTimestamp >= closeTime || msg.EventTimestamp < rt.MinWatermark() { return nil } state, err := flow.OpenExternalState(rt, joinStateName, msg) if err != nil { return err } var window joinState if _, err := state.ConvertTo(&window); err != nil { return err } if hit != nil { window.HitPayload = &hit.HitPayload } else if action.IsClick { window.ClickTime = &action.ActionTime } else { window.ShowTime = &action.ActionTime } if err := state.ConvertFrom(&window); err != nil { return err } out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime}) return nil } ``` ### `OnTimer` {#on-timer} Закрывает окно: публикует join-результат, если пришли обе стороны, и в любом случае очищает стейт: ```go func (*joinFunction) OnTimer( ctx context.Context, rt flow.Runtime, timer flow.Timer, out flow.OutputCollector, ) error { if timer.StreamID != timerStream { return fmt.Errorf("unhandled timer stream %q", timer.StreamID) } state, err := flow.OpenExternalState(rt, joinStateName, timer) if err != nil { return err } var window joinState if _, err := state.ConvertTo(&window); err != nil { return err } if window.ShowTime != nil && *window.ShowTime != 0 && window.HitPayload != nil { var key joinKey if err := timer.Key.ConvertTo(&key); err != nil { return err } joined, err := joinedAction(rt, key, window) if err != nil { return err } out.AddMessage(joined) } return state.Clear() } ``` ## Ключевые паттерны {#key-patterns} - Внешний стейт через `flow.OpenExternalState(rt, joinStateName, msg)`: `ConvertTo` читает окно в `joinState`, сообщение меняет нужное поле структуры, а `ConvertFrom` сохраняет окно. - Таймер с двумя таймстемпами: `out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime})` — `TriggerTimestamp` задаёт, когда окно закроется, `EventTimestamp` привязывает таймер к событию. Таймер взводит каждое сообщение, а не только `hit`: действие может прийти раньше показа, к которому относится, и окно всё равно надо закрыть. - Фильтрация запоздавших данных: `msg.EventTimestamp < rt.MinWatermark()` отбрасывает сообщения из уже закрытых окон — иначе для одного показа опубликовался бы второй join-результат. Подробнее о [вотермарках](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timestamps-and-watermarks). - Параметры компьютейшена: `rt.Parameters().Get(waitForActionsParameter, &spelled)` читает длину окна из спеки, а не зашивает её в код. - Обработка нескольких входных стримов: ветвление по `msg.StreamID` разводит логику для `hit` и `action`, а неизвестный стрим возвращает ошибку вместо молчаливого пропуска. - Ключ вместо стейта: `timer.Key` несёт `hit_id` и `hit_time` из `group_by_schema`, поэтому в окне хранятся только присоединяемые данные — дублировать в нём ключ не нужно. ## См. также - [Быстрый старт (Go)](https://ytsaurus.tech/docs/ru/flow/go/getting-started.md) - [External State (Go)](https://ytsaurus.tech/docs/ru/flow/go/external-state.md) - [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Вотермарки](https://ytsaurus.tech/docs/ru/flow/concepts/watermarks.md) - [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/wait_click_join.md)