--- metadata: - name: generator content: Diplodoc Platform v5.50.6 alternate: - https://ytsaurus.tech/docs/en/flow/java/examples/url_downloader.md - https://ytsaurus.tech/docs/ru/flow/java/examples/url_downloader.md --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt # URL Downloader в YTsaurus Flow (Java) [Пайплайн](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) группирует входящие URL по хосту, накапливает их во внутреннем [стейте](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state) и обрабатывает пакетами по таймеру: через 5 секунд после постановки URL в очередь срабатывает таймер, который эмитирует результаты в выходной [стрим](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation). [Исходный код (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/url_downloader) [Исходный код (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/url_downloader) ## Компоненты ### UrlDownloadFunction Основная процессная функция, реализующая логику накопления и обработки URL. Метод `onMessage` добавляет URL в внутренний стейт хоста и устанавливает таймер; метод `onTimer` обрабатывает накопленные URL и эмитирует результаты: {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.java' lang='java' lines='[BEGIN on_message]-[END on_message]' keep-indents %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.kt' lang='kotlin' lines='[BEGIN on_message]-[END on_message]' keep-indents %} {% endlist %} {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.java' lang='java' lines='[BEGIN on_timer]-[END on_timer]' keep-indents %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.kt' lang='kotlin' lines='[BEGIN on_timer]-[END on_timer]' keep-indents %} {% endlist %} ### HostState Модель внутреннего стейта, сериализуемая в YSON. Хранит имя хоста и список URL, ожидающих обработки: {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/model/HostState.java' lang='java' lines='[BEGIN host_state]-[END host_state]' keep-indents %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/model/HostState.kt' lang='kotlin' lines='[BEGIN host_state]-[END host_state]' keep-indents %} {% endlist %} ### Регистрация компьютейшена и стримов Компьютейшен `url_downloader` регистрируется аннотацией `@FlowComputation` на классе process-функции: {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.java' lang='java' lines='[BEGIN registration]-[END registration]' %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/UrlDownloadFunction.kt' lang='kotlin' lines='[BEGIN registration]-[END registration]' %} {% endlist %} Типизированные стримы объявляются через `ComputationProvider` (метод `getStreams()`): {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/UrlDownloaderComputationContext.java' lang='java' lines='[BEGIN stream_context]-[END stream_context]' keep-indents %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/UrlDownloaderComputationContext.kt' lang='kotlin' lines='[BEGIN stream_context]-[END stream_context]' keep-indents %} {% endlist %} ### PipelineMain Единственная точка входа (запускает пайплайн или обслуживает его как компаньон — по `YT_FLOW_MODE`): {% list tabs group=lang %} - Java {% code '/yt/yt/flow/examples/java/url_downloader/url_downloader/src/main/java/tech/ytsaurus/flow/examples/urldownloader/PipelineMain.java' lang='java' lines='[BEGIN main]-[END main]' keep-indents %} - Kotlin {% code '/yt/yt/flow/examples/kotlin/url_downloader/url_downloader/src/main/kotlin/tech/ytsaurus/flow/examples/urldownloader/PipelineMain.kt' lang='kotlin' lines='[BEGIN main]-[END main]' keep-indents %} {% endlist %} ## Ключевые паттерны - **Группировка по ключу** — стейт создаётся отдельно для каждого хоста; Flow автоматически направляет сообщения с одинаковым ключом в один экземпляр компьютейшена. - **Таймер на основе wall-clock времени** — `output.addTimer(System.currentTimeMillis() / 1000 + 5, 0L)` запускает обработку через 5 секунд после постановки URL в очередь. Несколько вызовов `addTimer` с одинаковым `triggerTimestamp` дедуплицируются. - **Пакетная обработка в `onTimer`** — все накопленные URL обрабатываются разом при срабатывании таймера, что снижает число обращений к downstream-сервисам. - **Очистка стейта** — после обработки стейт удаляется через `accessor.clear()`, предотвращая утечку памяти. - **YsonStateAccessor** — внутренний стейт сериализуется в YSON и хранится на стороне C++ воркера; Java-объект получается через `getOrDefault`. ## См. также - [Быстрый старт (Java)](https://ytsaurus.tech/docs/ru/flow/java/getting-started.md) - [Computation (Java)](https://ytsaurus.tech/docs/ru/flow/java/computation.md) - [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md) - [Аналогичный пример на C++](https://ytsaurus.tech/docs/ru/flow/cpp/examples/url_downloader.md)