--- metadata: - name: generator content: Diplodoc Platform v5.50.6 alternate: - https://ytsaurus.tech/docs/en/flow/cpp/examples/multiplexer.md - https://ytsaurus.tech/docs/ru/flow/cpp/examples/multiplexer.md --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt # Multiplexer в YTsaurus Flow (C++) Multiplexer — шаблон процесс-функции, который по входному ключу читает связанный с ним набор записей и отдаёт каждую запись отдельным выходным сообщением. Типичный пример: на вход поступает ключ X, а на выходе нужны все строки сортированной динамической таблицы, у которых X является префиксом ключа. Что обеспечивает базовый класс: - **равномерный прогресс** между несколькими активными ключами (один большой ключ не блокирует обработку остальных); - **обработка коллапсов** — если для уже обрабатываемого ключа прилетело новое входное сообщение, итерация перезапускается так, чтобы все строки были выданы с актуальной версией payload'а из входа. ## Как устроена итерация Базовый класс хранит состояние для каждого ключа с курсором (`Offset`) и при каждом срабатывании таймера вызывает у наследника `FetchBatch`, передавая текущий `startOffsetExclusive`. Наследник возвращает следующий курсор; если данные закончились — `nullopt`. При коллапсе (повторное входное сообщение для активного ключа) базовый класс запоминает текущую позицию (`InitialStartOffset`) и читает данные в два прохода: 1. дочитывает остаток от текущей позиции до конца (фаза 1); 2. возвращается к началу и доходит до сохранённой точки коллапса (фаза 2). Так гарантируется, что после коллапса будут выданы все строки с новой версией payload'а — включая те, которые уже были эмитены до коллапса со старой версией. ## Готовый класс: `TDynamicTableMultiplexerProcessFunction` {#dynamic-table-multiplexer} Покрывает типовой сценарий: есть сортированная динтаблица, а на вход прилетают сообщения с ключом и payload'ом. На каждое входное сообщение нужно выдать по одной строке для каждой записи таблицы с этим ключом. [Заголовок класса](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/multiplexer/dynamic_table_multiplexer_process_function.h) ### Параметры ```yson { "computation_class_name" = "NYT::NFlow::TProcessFunctionComputation"; "processing_function" = "TMyMultiplexerProcessFunction"; "processing_function_parameters" = { "table_path" = "//path/to/lookup_table"; }; } ``` Зарегистрируйте наследника с теми же типами статических и динамических параметров: ```cpp YT_FLOW_DEFINE_PROCESS_FUNCTION( TMyMultiplexerProcessFunction, TDynamicTableMultiplexerParameters, TDynamicMultiplexerParameters); ``` `table_path` — обязательный путь с указанием кластера. Перед первым чтением батча класс получает из схемы таблицы набор колонок для итерации (ключевые колонки после `group_by_schema`) и колонки данных, а затем кэширует их. `group_by_schema` компьютейшена должна совпадать с ведущими ключевыми колонками таблицы. После этого префикса в таблице должна быть хотя бы одна дополнительная сортировочная ключевая колонка. Базовый класс сохраняет схему дополнительных ключевых колонок вместе со смещением и сбрасывает смещение, если эта схема изменилась между запусками пайплайна. Изменения только в колонках данных не приводят к такому сбросу. Схема таблицы кэшируется на всё время работы экземпляра процесс-функции. В конфигурационном фрагменте выше показаны только поля процесс-функции. Для компьютейшена также нужны `group_by_schema`, входные и выходные потоки, поток таймеров типа `current_time`, зависимость таймера от потока таймеров и входного потока, а также `allow_timer_self_dependency = %true`. См. [реализацию функции](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/multiplexer/tests/pipeline/main.cpp) и её [спецификацию пайплайна](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/multiplexer/tests/pipeline/pipeline.yson). ### Что нужно реализовать Унаследуйтесь от класса и откройте его конструктор через `using TDynamicTableMultiplexerProcessFunction::TDynamicTableMultiplexerProcessFunction` либо определите конструктор, который принимает `TProcessFunctionContextPtr`. Переопределите `BuildOutputForRow`, чтобы построить выходное сообщение для одной выбранной строки. Метод получает входной ключ, строку и её схему, пользовательское состояние, коллектор выходных сообщений и контекст выполнения. Если требуется передать данные из входного сообщения в выходное, задайте свой `TUserState` и переопределите `OnInputMessage`. Иначе оставьте `TUserState` по умолчанию (`TEmptyMultiplexerUserState`) и не переопределяйте `OnInputMessage`. ```cpp void BuildOutputForRow( const TKey& key, const TPayload& rowPayload, const NTableClient::TTableSchemaPtr& rowSchema, TStateAccessor& userState, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context) override; ``` В полном примере по ссылкам выше используются вход `(key, payload)`, таблица поиска `[hash, key, secondary_key, region]` и выход `(key, secondary_key, region, payload)`. `rowPayload` — одна строка таблицы целиком (без group_by-колонок) как [`TPayload`](https://ytsaurus.tech/docs/ru/flow/cpp/state.md), `rowSchema` описывает её колонки (имена + типы). Колонки достаются по имени через `GetColumnValue` ([`payload.h`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/payload.h)). ### Динамические параметры Унаследованы от базового класса: - `timer_period` (по умолчанию 5 секунд) — как часто срабатывает таймер для ключа. - `batch_size` (по умолчанию 1000) — размер одного батча, передаётся в `LIMIT` запроса. В динамической спецификации компьютейшена укажите оба поля в `processing_function_parameters`: ```yson { "processing_function_parameters" = { "timer_period" = "10s"; "batch_size" = 1000; }; } ``` ## Базовый класс: `TMultiplexerProcessFunction` {#multiplexer-process-function} Если источник данных — не сортированная динамическая таблица, а, например, структура в памяти или пользовательский RPC-сервис, наследуйтесь напрямую от [`TMultiplexerProcessFunction`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/multiplexer/multiplexer_process_function.h) и реализуйте `FetchBatch`. Контракт `FetchBatch`: - **Offset** (`TKey`) — должен быть сравним и монотонно возрастать в рамках одной итерации. Базовый класс проверит монотонность и упадёт, если контракт нарушен. - **`startOffsetExclusive`** — читать данные строго **после** этой позиции. `nullopt` означает «с самого начала». - **`endOffsetInclusive`** — задаётся только в фазе 2 (после коллапса). Реализация должна не превышать эту границу — базовый класс это проверит. - **`limit`** — рекомендуемый максимум строк в одном батче. - **Возврат `nullopt`** — в текущем диапазоне данных больше нет. Базовый класс либо переключится в фазу 2, либо завершит итерацию. `OnInputMessage` вызывается для каждого входного сообщения ключа — как при появлении нового ключа, так и при коллапсе. Если реализации нужно различать эти случаи, храните в `TUserState` отдельный признак инициализации. Не полагайтесь на `userState.IsEmpty()`: состояние, состоящее из значений по умолчанию, останется пустым. Если сохранять данные из входного сообщения не требуется, не переопределяйте этот метод. ## См. также - [Процесс-функции (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md) - [Работа со стейтами (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/state.md)