Word Count в YTsaurus Flow (C++)

Простейший вводный пример на C++. Пайплайн читает текстовые сообщения из входной очереди, разбивает их на слова и подсчитывает количество вхождений каждого слова с помощью внешней таблицы стейта.

Исходный код

Обе единицы пользовательской логики написаны как process function (наследники IProcessFunction): сама функция лёгкая и тестируемая, а режим её исполнения выбирается встроенным Computation-адаптером в спеке через поле processing_function.

Компоненты пайплайна

TTextReadFunction

TTextReadFunction — process function (IProcessFunction), которую исполняет TProcessFunctionSourceComputation (source-адаптер). Она читает текстовые сообщения из входной очереди, разбивает текст на слова (по пробельным символам) и для каждого слова длиной не меньше min_word_length генерирует объект TWordMessage в выходной поток words. Параметр min_word_length читается в Init через initContext->GetParameters<TTextReaderParameters>() из блока processing_function_parameters спеки.

TProcessFunctionSourceComputation запускает функцию в Swift ordered-source режиме, поэтому выходные сообщения не сохраняются в YTsaurus — сохраняется только метаинформация, необходимая для детерминированной работы. Подробнее про режимы компьютейшенов можно прочитать в разделе Компьютейшены.

TWordCountFunction

TWordCountFunction — process function (IProcessFunction), которую исполняет TProcessFunctionComputation (transform-адаптер). Она использует TSimpleExternalStateManager для работы со стейтом. Для каждого входного слова она:

Получает текущее значение счетчика из внешней таблицы стейта
Увеличивает счетчик на 1
Записывает обновленное значение обратно

Подробнее про работу со стейтом — в разделе Стейт.

Типы сообщений

TWordMessage — наследник TYsonMessage. Содержит единственное поле Word. Регистрируется через макрос YT_FLOW_DEFINE_YSON_MESSAGE.

Структура пайплайна

Пайплайн состоит из двух компьютейшенов-адаптеров, соединенных потоком words:

source (входная очередь) → TProcessFunctionSourceComputation (processing_function = TTextReadFunction) → поток words
Поток words → TProcessFunctionComputation (processing_function = TWordCountFunction) → таблица стейта (word → count)

В спеке для счётчика указывается group_by_schema с хэшем от слова и самим словом — для корректного партиционирования. Параметры стейта (TSimpleExternalStateManager и путь к таблице) задаются декларативно в секции external_state_managers спеки Computation.

Функция main

В main выполняется:

NYT::NFlow::Initialize(argc, argv) — инициализация библиотеки Flow
Регистрация функций через YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction) и YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction)
TSimpleSpecBuilder — билдер для регистрации потоков. Через RegisterStream<TWordMessage>("words") регистрируется поток words с типом сообщений TWordMessage
TSimpleRunnerProgram — запуск пайплайна

Исходный код

TTextReadFunction

void TTextReadFunction::Init(const IRuntimeInitContextPtr& initContext)
{
    MinWordLength_ = initContext->GetParameters<TTextReaderParameters>()->MinWordLength;
}

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

TWordCountFunction

void TWordCountFunction::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitExternalStateClient(StateClient_, "/state");
}

void TWordCountFunction::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& /*output*/,
    const IRuntimeContextPtr& /*context*/)
{
    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();
}

См. также

Process function

Shuffle