--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md - https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.md - href: https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.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 # Быстрый старт в YTsaurus Flow (C++) В этом разделе пошагово описано, как реализовать свой первый [пайплайн](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) на C++ с использованием Flow. В качестве примера рассматривается задача подсчёта слов (word count): чтение текстовых сообщений из очереди, разбиение на слова и подсчёт вхождений каждого слова. ## Предварительные требования - Чекаут [репозитория](https://github.com/ytsaurus/ytsaurus/tree/main). - Настроенный `ya make` (сборочная система). - Ознакомление с [основными понятиями](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md) Flow. ## Пошаговое руководство ### 1. Определите типы сообщений {#define-messages} Для типизированной работы с сообщениями используйте `TYsonMessage` — специальный наследник `NYTree::TYsonStruct`. Каждый тип сообщения нужно зарегистрировать в глобальном реестре с помощью макроса `YT_FLOW_DEFINE_YSON_MESSAGE`. ```cpp #include struct TWordMessage : public TYsonMessage { std::string Word; REGISTER_YSON_STRUCT(TWordMessage); static void Register(TRegistrar registrar) { registrar.Parameter("word", &TThis::Word) .Default(); } }; YT_FLOW_DEFINE_YSON_MESSAGE(TWordMessage); ``` Подробнее о конвертации сообщений см. раздел [Process function (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md). ### 2. Подготовьте внешний стейт {#define-state} В этом примере счётчики хранятся во внешней таблице YTsaurus, а `TWordCountFunction` читает и обновляет их через `TMutableStateKeyClient`. Отдельный класс стейта не нужен: таблицу вы создадите на шаге 8, а `TSimpleExternalStateManager` укажете в спеке на шаге 6. Подробнее см. [Работа со стейтами (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/state.md). ### 3. Реализуйте process function для [Source](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#source) {#implement-source} Пользовательскую логику на C++ реализуйте только как [process function](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md). Для поэлементной обработки унаследуйтесь от `IProcessFunction` и реализуйте `ProcessMessage`: ```cpp class TTextReadFunction : public IProcessFunction { public: void ProcessMessage( const TInputMessageConstPtr& message, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context) override { auto text = GetColumnValue(message, "text"); for (const auto& word : StringSplitter(text).SplitBySet(" \t\n\r").SkipEmpty()) { auto wordMessage = New(); wordMessage->Word = word; output->AddMessage(context->ConvertToMessage(wordMessage)); } } }; YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction); ``` В спеке эту функцию исполняет встроенный `TProcessFunctionSourceComputation`. Он задаёт source-режим: выходные сообщения не материализуются в YT, сохраняется только метаинформация для восстановления. Подробнее см. [Process function](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md#how-it-works). ### 4. Реализуйте process function со стейтом {#implement-transform} Для обработки данных со стейтом также используйте `IProcessFunction`. Для работы с внешними стейтами используйте `TSimpleExternalStateManager`: ```cpp class TWordCountFunction : public IProcessFunction { public: void Init(const IRuntimeInitContextPtr& initContext) override { initContext->InitExternalStateClient(StateClient_, "/state"); } void ProcessMessage( const TInputMessageConstPtr& message, const IOutputCollectorPtr& /*output*/, const IRuntimeContextPtr& /*context*/) override { auto state = StateClient_.GetState(message->Key); i64 count = state->GetColumnValue>("count").value_or(0); TPayloadBuilder builder(state->Schema); builder.Set(count + 1, "count"); state->Payload = builder.Finish(); } private: TMutableStateKeyClient StateClient_; }; YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction); ``` Ключевые моменты: - `TMutableStateKeyClient` — типизированный клиент внешнего стейта; параметры менеджера задаются в спеке `Computation` (см. ниже), а не в собственных `TParameters`/`TDynamicParameters`. - `InitExternalStateClient(StateClient_, "/state")` — привязка клиента к external state manager'у с именем `"/state"`, объявленному в `external_state_managers` спеки. - В спеке функцию исполняет встроенный `TProcessFunctionComputation`, который задаёт transform-режим и exactly-once коммит стейта. ### 5. Напишите main.cpp {#write-main} Функция `main` связывает все компоненты вместе: ```cpp #include #include int main(int argc, const char** argv) { NYT::NFlow::Initialize(argc, argv); TSimpleSpecBuilder builder; builder.RegisterStream("words"); return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv); } ``` Здесь: - `Initialize(argc, argv)` — инициализация Flow runtime. - `TSimpleSpecBuilder` — билдер, в котором регистрируются все типизированные потоки. Он автоматически выводит схемы из зарегистрированных `TYsonMessage`. - `RegisterStream("words")` — регистрация потока `words` с типом сообщений `TWordMessage`. - `TSimpleRunnerProgram` — стандартный runner, который берёт на себя запуск и управление компьютейшенами. Полный исходный код примера: ```cpp #include #include #include #include #include using namespace NYT; using namespace NYT::NFlow; using namespace NYT::NFlow::NExample; //////////////////////////////////////////////////////////////////////////////// YT_FLOW_DEFINE_YSON_MESSAGE(TWordMessage); // Register the functions; the pipeline spec picks the hosting computation (the mode) and // names the function via the `processing_function` parameter. YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction, TTextReaderParameters); YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction); //////////////////////////////////////////////////////////////////////////////// int main(int argc, const char** argv) { NYT::NFlow::Initialize(argc, argv); TSimpleSpecBuilder builder; builder.RegisterStream("words"); return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv); } ``` ### 6. Опишите спеку пайплайна {#write-spec} Спека описывает топологию пайплайна в формате YSON. Пример для word count: ```yson { "spec" = { "computations" = { "reader" = { "computation_class_name" = "NYT::NFlow::TProcessFunctionSourceComputation"; "processing_function" = "NYT::NFlow::NExample::TTextReadFunction"; "output_stream_ids" = ["words"]; "source_streams" = { "queue" = { "source_class_name" = "NYT::NFlow::TQueueSource"; "parameters" = { "queue_path" = "//path/to/queue"; "consumer_path" = "//path/to/consumer"; }; }; }; }; "counter" = { "computation_class_name" = "NYT::NFlow::TProcessFunctionComputation"; "processing_function" = "NYT::NFlow::NExample::TWordCountFunction"; "input_stream_ids" = ["words"]; "output_stream_ids" = []; "group_by_schema" = [ {"name" = "hash"; "type" = "uint64"; "expression" = "farm_hash(word)";}; {"name" = "word"; "type" = "string";}; ]; "external_state_managers" = { "/state" = { "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager"; "parameters" = { "path" = "//path/to/state/table"; }; }; }; }; }; }; } ``` Заполнять секцию `streams` при использовании `TYsonMessage` и `TSimpleSpecBuilder` не требуется — информация о потоках будет выведена автоматически. Подробнее о формате спеки см. [Spec & DynamicSpec](https://ytsaurus.tech/docs/ru/flow/concepts/spec.md). ### 7. Соберите проект {#build} Фрагменты кода выше поясняют готовый пример Word Count, но не образуют отдельный проект. Соберите пример из репозитория; его зависимости заданы в [ya.make](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/word_count/ya.make): ```bash ya make yt/yt/flow/examples/cpp/word_count ``` Бинарь появится по пути `yt/yt/flow/examples/cpp/word_count/word_count`. Упрощённая спека из шага 6 не задаёт `min_word_length`, поэтому используется значение по умолчанию `0`; конфиг в репозитории задаёт `4`. Если вы перенесёте process function'ы в другое пространство имён, укажите их полные имена в полях `processing_function` спеки. ### 8. Создайте объекты в YT {#create-yt-objects} Перед запуском необходимо создать: - Входную очередь (если она ещё не существует). - Консьюмера входной очереди, зарегистрированного на ней. - Таблицу стейтов (для `ExternalState`). Как создать очередь, консьюмера и таблицу стейтов, показано в разделе [Подготовьте пайплайн](https://ytsaurus.tech/docs/ru/flow/quickstart.md#prepare) быстрого старта. C++-раннер при первом запуске создаёт пайплайн и его внутренние таблицы; если хотите создать их заранее, см. [Создание пайплайна](https://ytsaurus.tech/docs/ru/flow/concepts/pipeline-object.md#create). ### 9. Запустите и протестируйте {#run-and-test} Сохраните YSON из шага 6 в ``. Замените в нём пути к очереди, консьюмеру и внешней таблице стейтов, затем добавьте `cluster_url`, путь `path` к пайплайну и секцию `vanilla` с `enable = %true`, параметром `pool` с именем вашего пула и `worker = {"count" = 1;}`, как показано в разделе [Подготовьте пайплайн](https://ytsaurus.tech/docs/ru/flow/quickstart.md#prepare). Запустите собранный бинарь с `--config `, как в разделе [Проверьте и запустите](https://ytsaurus.tech/docs/ru/flow/quickstart.md#launch). Проверьте состояние командой `yt --proxy flow get-pipeline-state `, где `` — значение `path` из конфига; CLI должен вывести `working`. Перед выкаткой новой версии прочитайте [Правила выкатки и изменения таблиц](https://ytsaurus.tech/docs/ru/flow/devops/vanilla/releases.md#release-and-configure-basic-rules); команды управления пайплайном приведены в разделе [Операции с пайплайном](https://ytsaurus.tech/docs/ru/flow/devops/vanilla/pipeline-operations.md). ## См. также - [Process function (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md) - [Режимы Computation (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/computation.md) - [Работа со стейтами (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/state.md) - [Watermarks](https://ytsaurus.tech/docs/ru/flow/concepts/watermarks.md) - [Timers](https://ytsaurus.tech/docs/ru/flow/concepts/timers.md) - [Spec & DynamicSpec](https://ytsaurus.tech/docs/ru/flow/concepts/spec.md)