Proto Parser в YTsaurus Flow (C++)

Пример показывает TProtoParsingProcessFunctionBase<TProto> — базовый класс для process functions, который берёт на себя разбор Protobuf-сообщений. Пайплайн читает сериализованные записи лога из очереди, парсит их без ручного вызова ParseFromStringOrThrow и ведёт стейт — счётчик разобранных записей каждого уровня.

Исходный код

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

TProtoLogParserFunction

TProtoLogParserFunction наследуется от TProtoParsingProcessFunctionBase<TLogRecordProto>, где TLogRecordProto — Protobuf-сообщение с полями level и text. Базовый класс читает колонку входного сообщения, имя которой задано параметром data_column (по умолчанию "data"), разбирает её в TLogRecordProto и вызывает один из пользовательских методов:

ProcessProto(...) — при успешном разборе получает стейт по ключу сообщения, увеличивает счётчик уровня, собирает TLogRecordMessage и отправляет его в поток records;
ProcessUnparsed(...) — если колонка входного сообщения, имя которой задано параметром data_column, отсутствует (null) или Protobuf не разобрался. Реализация пустая, поэтому такие сообщения молча отбрасываются. Пустая, но присутствующая строка успешно разбирается в TLogRecordProto со значениями по умолчанию, поскольку в сообщении нет обязательных полей.

Process function регистрируется через YT_FLOW_DEFINE_PROCESS_FUNCTION. В спеке parser запускает её адаптер TProcessFunctionTransformOrderedSourceComputation, а поле processing_function содержит имя NYT::NFlow::NExample::TProtoLogParserFunction. При необходимости имя входной колонки можно передать через processing_function_parameters/data_column.

Выходной поток records сконфигурирован с прямым внешним TSyncQueueSink, поэтому отдельный sink-компьютейшен не требуется.

Стейт TLevelCountsState

TProtoLogParserFunction хранит TMutableStateKeyClient<TLevelCountsState> StateClient_ и инициализирует его в DoInit(const IRuntimeInitContextPtr&) через initContext->InitClient(StateClient_, "level_counts"). Метод ProcessProto получает стейт через StateClient_.GetState(message->Key). Стейт содержит record_counts — количество разобранных записей каждого уровня по партиции источника; текущее значение попадает в выходное сообщение как seen_at_level.

Адаптер TProcessFunctionTransformOrderedSourceComputation материализует выход и сохраняет стейт в той же транзакции эпохи, в которой продвигается смещение источника. Поэтому инкремент счётчика остаётся согласованным с чтением очереди при рестартах.

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

TLogRecordMessage — наследник TYsonMessage, зарегистрированный через YT_FLOW_DEFINE_YSON_MESSAGE, с полями:

level — уровень записи из TLogRecordProto;
text — текст записи из TLogRecordProto;
seen_at_level — число разобранных записей этого уровня на партиции источника, включая текущую.

Функция main

Регистрация process function и типов вынесена в библиотеку примера. В main выполняется инициализация Flow, регистрация потока records через TSimpleSpecBuilder и запуск TSimpleRunnerProgram.

Исходный код

TProtoLogParserFunction

#pragma once

#include <yt/yt/flow/examples/cpp/proto_parser/proto/log_record.pb.h>

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

#include <yt/yt/flow/library/cpp/parsers/proto.h>

namespace NYT::NFlow::NExample {

////////////////////////////////////////////////////////////////////////////////

struct TLogRecordMessage
    : public TYsonMessage
{
    std::string Level;
    std::string Text;
    i64 SeenAtLevel = 0;

    REGISTER_YSON_STRUCT(TLogRecordMessage);

    static void Register(TRegistrar registrar);
};

////////////////////////////////////////////////////////////////////////////////

struct TLevelCountsState
    : public NYTree::TYsonStruct
{
    THashMap<std::string, i64> RecordCounts;

    REGISTER_YSON_STRUCT(TLevelCountsState);

    static void Register(TRegistrar registrar);
};

////////////////////////////////////////////////////////////////////////////////

class TProtoLogParserFunction
    : public TProtoParsingProcessFunctionBase<
          TLogRecordProto,
          TProtoParsingProcessFunctionParameters,
          /*PropagateHookErrors*/ true>
{
protected:
    void DoInit(const IRuntimeInitContextPtr& initContext) override;

    void ProcessProto(
        const TInputMessageConstPtr& message,
        TLogRecordProto&& proto,
        const IOutputCollectorPtr& output,
        const IRuntimeContextPtr& context) override;

    void ProcessUnparsed(
        const TInputMessageConstPtr& message,
        TError error,
        const IOutputCollectorPtr& output,
        const IRuntimeContextPtr& context) override;

private:
    TMutableStateKeyClient<TLevelCountsState> StateClient_;
};

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NFlow::NExample

#include "proto_parser_function.h"

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

namespace NYT::NFlow::NExample {

////////////////////////////////////////////////////////////////////////////////

void TLogRecordMessage::Register(TRegistrar registrar)
{
    registrar.Parameter("level", &TThis::Level)
        .Default();
    registrar.Parameter("text", &TThis::Text)
        .Default();
    registrar.Parameter("seen_at_level", &TThis::SeenAtLevel)
        .Default();
}

void TLevelCountsState::Register(TRegistrar registrar)
{
    registrar.Parameter("record_counts", &TThis::RecordCounts)
        .Default();
}

////////////////////////////////////////////////////////////////////////////////

void TProtoLogParserFunction::DoInit(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitClient(StateClient_, "level_counts");
}

void TProtoLogParserFunction::ProcessProto(
    const TInputMessageConstPtr& message,
    TLogRecordProto&& proto,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto state = StateClient_.GetState(message->Key);
    auto record = New<TLogRecordMessage>();
    record->Level = proto.level();
    record->Text = proto.text();
    record->SeenAtLevel = ++state->RecordCounts[proto.level()];
    output->AddMessage(context->ConvertToMessage(record));
}

void TProtoLogParserFunction::ProcessUnparsed(
    const TInputMessageConstPtr& /*message*/,
    TError /*error*/,
    const IOutputCollectorPtr& /*output*/,
    const IRuntimeContextPtr& /*context*/)
{ }

////////////////////////////////////////////////////////////////////////////////

YT_FLOW_DEFINE_YSON_MESSAGE(TLogRecordMessage);
YT_FLOW_DEFINE_PROCESS_FUNCTION(TProtoLogParserFunction, TProtoParsingProcessFunctionParameters);

////////////////////////////////////////////////////////////////////////////////

} // namespace NYT::NFlow::NExample

См. также

Log Parser

Быстрый старт