---
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/en/flow/cpp/examples/proto_parser.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/cpp/examples/proto_parser.md -->
# Proto parser in YTsaurus Flow (C++)

This example shows `TProtoParsingProcessFunctionBase<TProto>`, a base class for [process functions](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md) that handles `Protobuf` message parsing. The [pipeline](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/proto_parser) reads serialized log records from a queue, parses them without calling `ParseFromStringOrThrow` manually, and maintains a state counter for parsed records of each level.

[Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/proto_parser)

## Pipeline components

### TProtoLogParserFunction

`TProtoLogParserFunction` inherits from `TProtoParsingProcessFunctionBase<TLogRecordProto>`, where `TLogRecordProto` is a `Protobuf` message with the `level` and `text` fields. The base class reads the input column named by the `data_column` parameter (the `"data"` column by default), parses it into `TLogRecordProto`, and invokes one of the user methods:

- `ProcessProto(...)`: after successful parsing, gets the state for the message key, increments the level counter, builds a `TLogRecordMessage`, and emits it to the `records` stream;
- `ProcessUnparsed(...)`: if the input column named by the `data_column` parameter is missing (`null`) or `Protobuf` parsing fails. The implementation is empty, so such messages are silently dropped. An empty but present string parses successfully into a `TLogRecordProto` with default values because the message has no required fields.

The process function is registered via `YT_FLOW_DEFINE_PROCESS_FUNCTION`. In the `parser` spec, `TProcessFunctionTransformOrderedSourceComputation` hosts it and `processing_function` contains `NYT::NFlow::NExample::TProtoLogParserFunction`. If necessary, the input column name can be set in `processing_function_parameters/data_column`.

The `records` output stream is configured with a direct external `TSyncQueueSink`, so no separate sink computation is required.

### The TLevelCountsState state

`TProtoLogParserFunction` stores a `TMutableStateKeyClient<TLevelCountsState> StateClient_` and initializes it in `DoInit(const IRuntimeInitContextPtr&)` via `initContext->InitClient(StateClient_, "level_counts")`. `ProcessProto` accesses the state through `StateClient_.GetState(message->Key)`. The state contains `record_counts`, the number of parsed records of each level per source partition; the current value is written to the output message as `seen_at_level`.

`TProcessFunctionTransformOrderedSourceComputation` materializes output and persists state in the same epoch transaction that advances the source offset. This keeps the counter consistent with queue consumption across restarts.

## Message types

`TLogRecordMessage` is a `TYsonMessage` subclass registered via `YT_FLOW_DEFINE_YSON_MESSAGE` with the following fields:

- `level`: the record level from `TLogRecordProto`;
- `text`: the record text from `TLogRecordProto`;
- `seen_at_level`: the number of parsed records of this level in the source partition, including the current record.

## The main function

The example library contains process-function and type registration. `main` initializes Flow, registers the `records` stream through `TSimpleSpecBuilder`, and starts `TSimpleRunnerProgram`.

## Source code

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

## See also

- [Getting started (C++)](https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md)
- [Process functions (C++)](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md)
- [Computation (C++)](https://ytsaurus.tech/docs/en/flow/cpp/computation.md)
- [Log parser](https://ytsaurus.tech/docs/en/flow/cpp/examples/log_parser.md)
<!-- endsource: en/_includes/flow/cpp/examples/proto_parser.md -->