Proto parser in YTsaurus Flow (C++)

This example shows TProtoParsingProcessFunctionBase<TProto>, a base class for process functions that handles Protobuf message parsing. The pipeline 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

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

#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

Proto parser in YTsaurus Flow (C++)

Pipeline components

TProtoLogParserFunction

The TLevelCountsState state

Message types

The main function

Source code

TProtoLogParserFunction

See also