Wait Click Join в YTsaurus Flow (Go)

Пример join-пайплайна, который объединяет стримы показов (hit) и действий (action) по общему ключу, используя таймеры и внешний стейт. Go-реализация аналогичного примера на C++.

Исходный код

Структура

Пайплайн состоит из трёх компьютейшенов:

hit_reader и action_reader — нативные сорсы (TSwiftPassthroughOrderedSourceComputation), объявленные прямо в спеке: они читают две очереди и публикуют стримы hit и action. Go-кода у них нет. Каждый сорс объявляет watermark_strategy, поэтому вотермарк пайплайна считается по времени события.
join (joinFunction) — transform-компьютейшен, который обслуживает компаньон.

Компьютейшен join работает так:

Получает сообщения из двух входных стримов: hit и action.
Накапливает окно одного показа во внешнем стейте /join-state.
Взводит таймер на момент закрытия окна ожидания.
По срабатыванию таймера публикует join-результат в стрим joined_action либо просто очищает стейт.

Оба входных стрима группируются по hit_id и hit_time, поэтому всё, что относится к одному показу, попадает в один ключ. Длину окна задаёт параметр компьютейшена wait_for_actions из спеки.

`main.go`

Точка входа: создание пайплайна, регистрация единственного компьютейшена и запуск.

Исходный код: main.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-а: функция обрабатывает два стрима и использует таймеры для оконной агрегации.

`OnMessage`

Отбрасывает запоздавшие сообщения, дописывает своё поле в окно показа и взводит таймер на закрытие окна:


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`

Закрывает окно: публикует join-результат, если пришли обе стороны, и в любом случае очищает стейт:


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()
}

Ключевые паттерны

Внешний стейт через flow.OpenExternalState(rt, joinStateName, msg): ConvertTo читает окно в joinState, сообщение меняет нужное поле структуры, а ConvertFrom сохраняет окно.
Таймер с двумя таймстемпами: out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime}) — TriggerTimestamp задаёт, когда окно закроется, EventTimestamp привязывает таймер к событию. Таймер взводит каждое сообщение, а не только hit: действие может прийти раньше показа, к которому относится, и окно всё равно надо закрыть.
Фильтрация запоздавших данных: msg.EventTimestamp < rt.MinWatermark() отбрасывает сообщения из уже закрытых окон — иначе для одного показа опубликовался бы второй join-результат. Подробнее о вотермарках.
Параметры компьютейшена: rt.Parameters().Get(waitForActionsParameter, &spelled) читает длину окна из спеки, а не зашивает её в код.
Обработка нескольких входных стримов: ветвление по msg.StreamID разводит логику для hit и action, а неизвестный стрим возвращает ошибку вместо молчаливого пропуска.
Ключ вместо стейта: timer.Key несёт hit_id и hit_time из group_by_schema, поэтому в окне хранятся только присоединяемые данные — дублировать в нём ключ не нужно.

См. также

Retryable Async Request

URL Downloader