Log parser in YTsaurus Flow (C++)
This example shows a process function in materialized ordered-source mode. The pipeline reads log lines from a queue and parses them while reading the source, without an intermediate passthrough computation. The built-in TProcessFunctionTransformOrderedSourceComputation executes the function. The example also shows reading from a Source and maintaining durable state that survives restarts (see State).
Pipeline components
TLogParserProcessFunction
The user logic is written as a process function: a subclass of IProcessFunction that doesn’t depend on the Computation object, so you can cover it with unit tests without a cluster (unittest). It’s executed by the built-in TProcessFunctionTransformOrderedSourceComputation adapter, which also sets the mode to ordered source (see the list of adapters).
In ProcessMessage(const TInputMessageConstPtr& message, const IOutputCollectorPtr& output, const IRuntimeContextPtr& context), the function reads the line column of the raw source message via GetColumnValue<std::string>(message, "line") and parses it with ParseLogLine into records of the form "level:text" separated by ;, discarding records without the : separator, with empty text, or with a level other than info, warning, and error. For each valid record, it updates the state through the StateClient_.GetState(message->Key) accessor (see State), builds a TLogRecordMessage, and emits it into the records output stream by calling output->AddMessage(context->ConvertToMessage(outputRecord)).
The transformation result, the records stream, is materialized in YTsaurus as in TTransformComputation, so the transformation has no determinism requirements: after a restart, Flow distributes the already materialized messages with the MessageId values previously assigned to them instead of recomputing them.
The Proto parser example uses the same hosting mode through a reusable process-function base: TProtoLogParserFunction inherits from TProtoParsingProcessFunctionBase<TLogRecordProto> and runs under TProcessFunctionTransformOrderedSourceComputation. The host adapter’s spec validator is the same as the computation base class’s: a non-empty group_by_schema, timers, key-visitor streams, and external_state_managers are rejected (for the full list of limitations).
The parser computation spec
Two fields of the parser computation spec link the function to the adapter (see Registration):
"computation_class_name" = "NYT::NFlow::TProcessFunctionTransformOrderedSourceComputation";
"processing_function" = "NYT::NFlow::NExample::TLogParserProcessFunction";
The remaining fields of the parser record describe the connections: source_streams.queue is a TQueueSource with the paths to the queue and the consumer, and sinks.queue is a direct external TSyncQueueSink for the records stream, so no separate sink computation is needed. For the full file, see pipeline.yson.
Message types
TLogRecordMessage is a subclass of TYsonMessage (a YSON structure registered via YT_FLOW_DEFINE_YSON_MESSAGE) with the following fields:
level: the record level (info,warning, orerror);text: the record text;worst_level_so_far: the highest severity level (info < warning < error) encountered in this source partition at the time of the record (see State).
State
TLogParserProcessFunction is stateful. It keeps the TWorstSeverityState state in the TMutableStateKeyClient<TWorstSeverityState> StateClient_ field (see Working with states (C++)). The TProcessFunctionTransformOrderedSourceComputation adapter calls Init(const IRuntimeInitContextPtr& initContext), where the client connects to the state via initContext->InitClient(StateClient_, WorstSeverityStateName) (the state name is worst_severity), and ProcessMessage, where the state is read through GetState(message->Key) and output records are converted into messages via context->ConvertToMessage(...).
The computation instance is bound to a single source partition, so all messages carry the same key and address the same state row: state->WorstSeverity = std::max(state->WorstSeverity, SeverityRank(record.Level)).
The framework synchronizes this state in the same epoch transaction as the advance of the source offset (see Computation); the function itself does nothing for this. Therefore, user state is exactly-once correct and doesn’t have to be idempotent under reprocessing: an ordinary, “naive” counter of processed records would be just as correct here. The example keeps a running maximum severity simply because it’s a natural aggregate for such a pipeline, not because it’s in any way safer than a counter.
The main function
The main function does the following:
NYT::NFlow::Initialize(argc, argv): initializes the Flow library.TSimpleSpecBuilder: the builder for registering streams.RegisterStream<TLogRecordMessage>("records")registers therecordsstream with theTLogRecordMessagemessage type.TSimpleRunnerProgram(std::move(builder)).Run(argc, argv): starts the pipeline.
You don’t need to register the function in main: the YT_FLOW_DEFINE_PROCESS_FUNCTION(TLogParserProcessFunction) macro is placed at file level in lib/log_parser_process_function.cpp, and the file itself is linked into the library as GLOBAL, so the registry entry appears when the binary is initialized.
Source code
TLogParserProcessFunction
#pragma once
#include "log_line_parser.h"
#include <yt/yt/flow/library/cpp/common/process_function.h>
#include <yt/yt/flow/library/cpp/common/runtime_init_context.h>
#include <yt/yt/flow/library/cpp/common/state_client.h>
#include <yt/yt/flow/library/cpp/common/yson_message.h>
#include <string>
namespace NYT::NFlow::NExample {
////////////////////////////////////////////////////////////////////////////////
struct TLogRecordMessage
: public TYsonMessage
{
std::string Level;
std::string Text;
std::string WorstLevelSoFar;
REGISTER_YSON_STRUCT(TLogRecordMessage);
static void Register(TRegistrar registrar)
{
registrar.Parameter("level", &TThis::Level)
.Default();
registrar.Parameter("text", &TThis::Text)
.Default();
registrar.Parameter("worst_level_so_far", &TThis::WorstLevelSoFar)
.Default();
}
};
////////////////////////////////////////////////////////////////////////////////
struct TWorstSeverityState
: public NYTree::TYsonStruct
{
int WorstSeverity = -1;
REGISTER_YSON_STRUCT(TWorstSeverityState);
static void Register(TRegistrar registrar)
{
registrar.Parameter("worst_severity", &TThis::WorstSeverity)
.Default(-1);
}
};
////////////////////////////////////////////////////////////////////////////////
inline constexpr TStringBuf WorstSeverityStateName = "worst_severity";
class TLogParserProcessFunction
: public IProcessFunction
{
public:
void Init(const IRuntimeInitContextPtr& initContext) override;
void ProcessMessage(
const TInputMessageConstPtr& message,
const IOutputCollectorPtr& output,
const IRuntimeContextPtr& context) override;
private:
TMutableStateKeyClient<TWorstSeverityState> StateClient_;
};
////////////////////////////////////////////////////////////////////////////////
} // namespace NYT::NFlow::NExample
#include "log_parser_process_function.h"
#include <yt/yt/flow/library/cpp/common/input_context.h>
#include <yt/yt/flow/library/cpp/common/registry.h>
#include <yt/yt/flow/library/cpp/common/runtime_context.h>
#include <algorithm>
namespace NYT::NFlow::NExample {
////////////////////////////////////////////////////////////////////////////////
void TLogParserProcessFunction::Init(const IRuntimeInitContextPtr& initContext)
{
initContext->InitClient(StateClient_, WorstSeverityStateName);
}
void TLogParserProcessFunction::ProcessMessage(
const TInputMessageConstPtr& message,
const IOutputCollectorPtr& output,
const IRuntimeContextPtr& context)
{
auto state = StateClient_.GetState(message->Key);
for (const auto& record : ParseLogLine(GetColumnValue<std::string>(message, "line"))) {
state->WorstSeverity = std::max(state->WorstSeverity, SeverityRank(record.Level));
auto outputRecord = New<TLogRecordMessage>();
outputRecord->Level = record.Level;
outputRecord->Text = record.Text;
outputRecord->WorstLevelSoFar = SeverityName(state->WorstSeverity);
output->AddMessage(context->ConvertToMessage(outputRecord));
}
}
////////////////////////////////////////////////////////////////////////////////
YT_FLOW_DEFINE_YSON_MESSAGE(TLogRecordMessage);
YT_FLOW_DEFINE_PROCESS_FUNCTION(TLogParserProcessFunction);
////////////////////////////////////////////////////////////////////////////////
} // namespace NYT::NFlow::NExample
ParseLogLine
#include "log_line_parser.h"
#include <util/string/split.h>
namespace NYT::NFlow::NExample {
////////////////////////////////////////////////////////////////////////////////
int SeverityRank(TStringBuf level)
{
if (level == "info") {
return 0;
}
if (level == "warning") {
return 1;
}
if (level == "error") {
return 2;
}
return -1;
}
std::string SeverityName(int rank)
{
switch (rank) {
case 0:
return "info";
case 1:
return "warning";
case 2:
return "error";
default:
return {};
}
}
std::vector<TLogRecord> ParseLogLine(const std::string& line)
{
std::vector<TLogRecord> records;
for (const auto& entry : StringSplitter(line).Split(';').SkipEmpty()) {
TStringBuf token = entry.Token();
auto separator = token.find(':');
if (separator == TStringBuf::npos) {
continue;
}
auto level = token.substr(0, separator);
auto text = token.substr(separator + 1);
if (text.empty() || SeverityRank(level) < 0) {
continue;
}
records.push_back({.Level = std::string(level), .Text = std::string(text)});
}
return records;
}
////////////////////////////////////////////////////////////////////////////////
} // namespace NYT::NFlow::NExample