---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/examples/proto_parser.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/examples/proto_parser.md
  - href: https://ytsaurus.tech/docs/ru/flow/cpp/examples/proto_parser.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

<!-- source: ru/_includes/flow/cpp/examples/proto_parser.md -->
# Proto Parser в YTsaurus Flow (C++)

Пример показывает `TProtoParsingProcessFunctionBase<TProto>` — базовый класс для [process functions](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md), который берёт на себя разбор `Protobuf`-сообщений. [Пайплайн](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/proto_parser) читает сериализованные записи лога из очереди, парсит их без ручного вызова `ParseFromStringOrThrow` и ведёт стейт — счётчик разобранных записей каждого уровня.

[Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/proto_parser)

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

### 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

```cpp
#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
```

```cpp
#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
```

## См. также

- [Быстрый старт (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.md)
- [Process functions (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/process-functions.md)
- [Computation (C++)](https://ytsaurus.tech/docs/ru/flow/cpp/computation.md)
- [Log Parser](https://ytsaurus.tech/docs/ru/flow/cpp/examples/log_parser.md)
<!-- endsource: ru/_includes/flow/cpp/examples/proto_parser.md -->