URL Downloader в YTsaurus Flow (Go)

Пример пайплайна, который группирует входящие URL по хосту и обрабатывает их пакетно по таймеру. Показывает типичный паттерн «накопить в стейте → обработать по таймеру → очистить стейт».

Исходный код

Структура

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

url_reader — нативный сорс (TSwiftPassthroughOrderedSourceComputation), объявленный прямо в спеке: он читает очередь и публикует сообщения с полями host и url в стрим urls. Go-кода у него нет.
url_downloader (urlDownloadFunction) — transform-компьютейшен, который обслуживает компаньон.

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

Принимает сообщения из стрима urls.
Дописывает URL в батч своего хоста во внутреннем YSON-стейте host-state.
Взводит таймер на flushDelay (5 секунд) вперёд при каждом новом URL.
По срабатыванию таймера обрабатывает весь накопленный батч, публикует результаты в стрим processed_urls и очищает стейт.

Сообщения группируются по хосту (group_by_schema с farm_hash(host) и host), поэтому стейт обрабатываемого ключа — это батч ровно одного хоста.

`main.go`

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

Исходный код: main.go


type urlMessage struct {
	flow.YSONMessage
	Host string `yson:"host"`
	URL  string `yson:"url"`
}

type processedURLMessage struct {
	flow.YSONMessage
	Host string `yson:"host"`
	URL  string `yson:"url"`
	Data string `yson:"data"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[urlMessage]("urls"),
		flow.NewYSONStream[processedURLMessage](outputStreamID),
	)
	pipeline.Add(flow.NewRowComputation("url_downloader", &urlDownloadFunction{}))

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "url_downloader: %v\n", err)
		os.Exit(1)
	}
}

`url_download_function.go`

Значение стейта — обычная Go-структура с YSON-тегами: в ней лежат имя хоста и список ещё не обработанных URL.


type hostState struct {
	Host        string   `yson:"host"`
	PendingURLs []string `yson:"pending_urls"`
}

flow.RowFunction реализует оба обработчика: OnMessage накапливает URL в стейте и взводит таймер, OnTimer обрабатывает батч целиком.


type urlDownloadFunction struct{}

var (
	_ flow.RowFunction      = (*urlDownloadFunction)(nil)
	_ flow.RowTimerFunction = (*urlDownloadFunction)(nil)
)

func (*urlDownloadFunction) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var input urlMessage
	if err := msg.ConvertTo(&input); err != nil {
		return err
	}

	state, err := flow.OpenYSONState[hostState](rt, hostStateName, msg)
	if err != nil {
		return err
	}

	fresh := state.Empty()
	batch := state.Value()
	if fresh {
		batch.Host = input.Host
	}
	batch.PendingURLs = append(batch.PendingURLs, input.URL)

	out.AddTimer(flow.TimerRequest{TriggerTimestamp: uint64(time.Now().Add(flushDelay).Unix())})
	return nil
}

func (*urlDownloadFunction) OnTimer(
	ctx context.Context,
	rt flow.Runtime,
	timer flow.Timer,
	out flow.OutputCollector,
) error {
	state, err := flow.OpenYSONState[hostState](rt, hostStateName, timer)
	if err != nil {
		return err
	}

	if state.Empty() {
		return nil
	}
	batch := state.Value()
	if len(batch.PendingURLs) == 0 {
		state.Clear()
		return nil
	}

	for _, url := range batch.PendingURLs {
		processed := flow.NewYSONMessage[processedURLMessage](outputStreamID)
		processed.Host = batch.Host
		processed.URL = url
		processed.Data = processURL(url)
		encoded, err := flow.ConvertFrom(rt, processed)
		if err != nil {
			return err
		}
		out.AddMessage(encoded)
	}

	state.Clear()
	return nil
}

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

Пакетная обработка по таймеру: out.AddTimer(flow.TimerRequest{TriggerTimestamp: ...}) в OnMessage и вся обработка в OnTimer — стандартный способ собирать события во временное окно. Воркер хранит один таймер на ключ, поэтому всплеск сообщений схлопывается в одно срабатывание, а не в срабатывание на каждое сообщение.
Внутренний YSON-стейт через flow.OpenYSONState[hostState](rt, hostStateName, msg): Value() возвращает изменяемый батч, а Clear() очищает его после обработки. Имя стейта (host-state) совпадает с именем из parameters.internal_states компьютейшена в спеке.
Ключ стейта по хосту задаётся group_by_schema из спеки, так что каждая партиция компьютейшена обрабатывает URL одного хоста изолированно.
Каждый результат создаётся как processedURLMessage и один раз преобразуется в выходное сообщение через flow.ConvertFrom.
Очистка стейта на всех путях: OnTimer вызывает state.Clear() и когда батч пуст, и когда он обработан, — сработавший таймер всегда оставляет ключ чистым, и пришедшие после этого URL образуют новый батч.

См. также

Wait Click Join

Быстрый старт