Быстрый старт в YTsaurus Flow (C++)

В этом разделе пошагово описано, как реализовать свой первый пайплайн на C++ с использованием Flow. В качестве примера рассматривается задача подсчёта слов (word count): чтение текстовых сообщений из очереди, разбиение на слова и подсчёт вхождений каждого слова.

Предварительные требования

Пошаговое руководство

1. Определите типы сообщений

Для типизированной работы с сообщениями используйте TYsonMessage — специальный наследник NYTree::TYsonStruct. Каждый тип сообщения нужно зарегистрировать в глобальном реестре с помощью макроса YT_FLOW_DEFINE_YSON_MESSAGE.

#include <yt/yt/flow/library/cpp/common/registry.h>

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++).

2. Подготовьте внешний стейт

В этом примере счётчики хранятся во внешней таблице YTsaurus, а TWordCountFunction читает и обновляет их через TMutableStateKeyClient<TSimpleExternalState>. Отдельный класс стейта не нужен: таблицу вы создадите на шаге 8, а TSimpleExternalStateManager укажете в спеке на шаге 6. Подробнее см. Работа со стейтами (C++).

3. Реализуйте process function для Source

Пользовательскую логику на C++ реализуйте только как process function. Для поэлементной обработки унаследуйтесь от IProcessFunction и реализуйте ProcessMessage:

class TTextReadFunction
    : public IProcessFunction
{
public:
    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& output,
        const IRuntimeContextPtr& context) override
    {
        auto text = GetColumnValue<std::string>(message, "text");
        for (const auto& word : StringSplitter(text).SplitBySet(" \t\n\r").SkipEmpty()) {
            auto wordMessage = New<TWordMessage>();
            wordMessage->Word = word;
            output->AddMessage(context->ConvertToMessage(wordMessage));
        }
    }
};

YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction);

В спеке эту функцию исполняет встроенный TProcessFunctionSourceComputation. Он задаёт source-режим: выходные сообщения не материализуются в YT, сохраняется только метаинформация для восстановления. Подробнее см. Process function.

4. Реализуйте process function со стейтом

Для обработки данных со стейтом также используйте IProcessFunction. Для работы с внешними стейтами используйте TSimpleExternalStateManager:

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<std::optional<i64>>("count").value_or(0);
        TPayloadBuilder builder(state->Schema);
        builder.Set(count + 1, "count");
        state->Payload = builder.Finish();
    }

private:
    TMutableStateKeyClient<TSimpleExternalState> StateClient_;
};

YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction);

Ключевые моменты:

  • TMutableStateKeyClient<TState> — типизированный клиент внешнего стейта; параметры менеджера задаются в спеке Computation (см. ниже), а не в собственных TParameters/TDynamicParameters.
  • InitExternalStateClient(StateClient_, "/state") — привязка клиента к external state manager'у с именем "/state", объявленному в external_state_managers спеки.
  • В спеке функцию исполняет встроенный TProcessFunctionComputation, который задаёт transform-режим и exactly-once коммит стейта.

5. Напишите main.cpp

Функция main связывает все компоненты вместе:

#include <yt/yt/flow/library/cpp/runner/init.h>
#include <yt/yt/flow/library/cpp/runner/simple_runner_program.h>

int main(int argc, const char** argv)
{
    NYT::NFlow::Initialize(argc, argv);
    TSimpleSpecBuilder builder;
    builder.RegisterStream<TWordMessage>("words");
    return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv);
}

Здесь:

  • Initialize(argc, argv) — инициализация Flow runtime.
  • TSimpleSpecBuilder — билдер, в котором регистрируются все типизированные потоки. Он автоматически выводит схемы из зарегистрированных TYsonMessage.
  • RegisterStream<TWordMessage>("words") — регистрация потока words с типом сообщений TWordMessage.
  • TSimpleRunnerProgram — стандартный runner, который берёт на себя запуск и управление компьютейшенами.

Полный исходный код примера:

#include <yt/yt/flow/examples/cpp/word_count/lib/word_count_functions.h>

#include <yt/yt/flow/library/cpp/common/registry.h>
#include <yt/yt/flow/library/cpp/common/spec.h>


#include <yt/yt/flow/library/cpp/runner/init.h>
#include <yt/yt/flow/library/cpp/runner/simple_runner_program.h>

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<TWordMessage>("words");
    return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv);
}

6. Опишите спеку пайплайна

Спека описывает топологию пайплайна в формате YSON. Пример для word count:

{
    "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" = "<cluster=cluster_name>//path/to/queue";
                            "consumer_path" = "<cluster=cluster_name>//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.

7. Соберите проект

Фрагменты кода выше поясняют готовый пример Word Count, но не образуют отдельный проект. Соберите пример из репозитория; его зависимости заданы в ya.make:

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

Перед запуском необходимо создать:

  • Входную очередь (если она ещё не существует).
  • Консьюмера входной очереди, зарегистрированного на ней.
  • Таблицу стейтов (для ExternalState).

Как создать очередь, консьюмера и таблицу стейтов, показано в разделе Подготовьте пайплайн быстрого старта. C++-раннер при первом запуске создаёт пайплайн и его внутренние таблицы; если хотите создать их заранее, см. Создание пайплайна.

9. Запустите и протестируйте

Сохраните YSON из шага 6 в <config.yson>. Замените в нём пути к очереди, консьюмеру и внешней таблице стейтов, затем добавьте cluster_url, путь path к пайплайну и секцию vanilla с enable = %true, параметром pool с именем вашего пула и worker = {"count" = 1;}, как показано в разделе Подготовьте пайплайн. Запустите собранный бинарь с --config <config.yson>, как в разделе Проверьте и запустите. Проверьте состояние командой yt --proxy <cluster> flow get-pipeline-state <pipeline_path>, где <pipeline_path> — значение path из конфига; CLI должен вывести working.

Перед выкаткой новой версии прочитайте Правила выкатки и изменения таблиц; команды управления пайплайном приведены в разделе Операции с пайплайном.

См. также

Предыдущая
Spec, DynamicSpec и Config
Следующая
Computation