--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/go/examples/url_downloader.md - https://ytsaurus.tech/docs/ru/flow/go/examples/url_downloader.md - href: https://ytsaurus.tech/docs/ru/flow/go/examples/url_downloader.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 # URL Downloader в YTsaurus Flow (Go) Пример [пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline), который группирует входящие URL по хосту и обрабатывает их пакетно по [таймеру](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#timer). Показывает типичный паттерн «накопить в [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) → обработать по таймеру → очистить стейт». [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/url_downloader) ## Структура {#structure} Пайплайн состоит из двух [компьютейшенов](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation): - `url_reader` — нативный сорс (`TSwiftPassthroughOrderedSourceComputation`), объявленный прямо в [спеке](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec): он читает очередь и публикует сообщения с полями `host` и `url` в [стрим](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) `urls`. Go-кода у него нет. - `url_downloader` (`urlDownloadFunction`) — transform-компьютейшен, который обслуживает компаньон. Компьютейшен `url_downloader` работает так: 1. Принимает сообщения из стрима `urls`. 2. Дописывает URL в батч своего хоста во внутреннем YSON-стейте `host-state`. 3. Взводит таймер на `flushDelay` (5 секунд) вперёд при каждом новом URL. 4. По срабатыванию таймера обрабатывает весь накопленный батч, публикует результаты в стрим `processed_urls` и очищает стейт. Сообщения группируются по хосту (`group_by_schema` с `farm_hash(host)` и `host`), поэтому стейт обрабатываемого ключа — это батч ровно одного хоста. ## `main.go` {#main-go} Точка входа: создание пайплайна, регистрация единственного компьютейшена и запуск. [Исходный код: `main.go`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/url_downloader/main.go) ```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` {#url-download-function-go} Значение стейта — обычная Go-структура с YSON-тегами: в ней лежат имя хоста и список ещё не обработанных URL. ```go type hostState struct { Host string `yson:"host"` PendingURLs []string `yson:"pending_urls"` } ``` `flow.RowFunction` реализует оба обработчика: `OnMessage` накапливает URL в стейте и взводит таймер, `OnTimer` обрабатывает батч целиком. ```go 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 } ``` ## Ключевые паттерны {#key-patterns} - Пакетная обработка по таймеру: `out.AddTimer(flow.TimerRequest{TriggerTimestamp: ...})` в `OnMessage` и вся обработка в `OnTimer` — стандартный способ собирать события во временное окно. Воркер хранит один таймер на ключ, поэтому всплеск сообщений схлопывается в одно срабатывание, а не в срабатывание на каждое сообщение. - Внутренний YSON-стейт через `flow.OpenYSONState[hostState](rt, hostStateName, msg)`: `Value()` возвращает изменяемый батч, а `Clear()` очищает его после обработки. Имя стейта (`host-state`) совпадает с именем из `parameters.internal_states` компьютейшена в спеке. - Ключ стейта по хосту задаётся `group_by_schema` из [спеки](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec), так что каждая партиция компьютейшена обрабатывает URL одного хоста изолированно. - Каждый результат создаётся как `processedURLMessage` и один раз преобразуется в выходное сообщение через `flow.ConvertFrom`. - Очистка стейта на всех путях: `OnTimer` вызывает `state.Clear()` и когда батч пуст, и когда он обработан, — сработавший таймер всегда оставляет ключ чистым, и пришедшие после этого URL образуют новый батч. ## См. также - [Быстрый старт (Go)](https://ytsaurus.tech/docs/ru/flow/go/getting-started.md) - [Computation (Go)](https://ytsaurus.tech/docs/ru/flow/go/computation.md) - [Internal State (Go)](https://ytsaurus.tech/docs/ru/flow/go/internal-state.md) - [Таймеры](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/url_downloader.md)