---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/getting-started.md
  - href: https://ytsaurus.tech/docs/en/flow/cpp/getting-started.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/getting-started.md -->
# Quick start with YTsaurus Flow (C++)

In this section, you’ll learn step by step how to implement your first [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) in C++ using Flow. As an example, we’ll walk through a word count task: reading text messages from a queue, splitting them into words, and counting the occurrences of each word.

## Prerequisites

- Check out the [repository](https://github.com/ytsaurus/ytsaurus/tree/main).
- Set up `ya make` (the build system).
- Familiarize yourself with the [basic concepts](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md) of Flow.

## Step-by-step guide

### 1. Define message types {#define-messages}

To work with messages in a type-safe way, use `TYsonMessage`, a special subclass of `NYTree::TYsonStruct`. You need to register each message type in the global registry using the `YT_FLOW_DEFINE_YSON_MESSAGE` macro.

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

struct TWordMessage
    : public TYsonMessage
{
    std::string Word;

    REGISTER_YSON_STRUCT(TWordMessage);

    static void Register(TRegistrar registrar)
    {
        registrar.Parameter("word", &TThis::Word)
            .Default();
    }
};

YT_FLOW_DEFINE_YSON_MESSAGE(TWordMessage);
```

For more details on message conversion, see [Process functions (C++)](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md).

### 2. Prepare external state {#define-state}

In this example, the counters live in an external YTsaurus table, and `TWordCountFunction` reads and updates them through `TMutableStateKeyClient<TSimpleExternalState>`. You don't need a separate state class: create the table in step 8 and configure `TSimpleExternalStateManager` in the step 6 spec. For details, see [Working with states (C++)](https://ytsaurus.tech/docs/en/flow/cpp/state.md).
### 3. Implement a process function for the [Source](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#source) {#implement-source}

Implement C++ user logic only as a [process function](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md). For element-wise processing, inherit from `IProcessFunction` and implement `ProcessMessage`:

```cpp
class TTextReadFunction
    : public IProcessFunction
{
public:
    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& output,
        const IRuntimeContextPtr& context) override
    {
        auto text = GetColumnValue<std::string>(message, "text");
        for (const auto& word : StringSplitter(text).SplitBySet(" \t\n\r").SkipEmpty()) {
            auto wordMessage = New<TWordMessage>();
            wordMessage->Word = word;
            output->AddMessage(context->ConvertToMessage(wordMessage));
        }
    }
};

YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction);
```

In the spec, the built-in `TProcessFunctionSourceComputation` executes this function. It selects source mode: output messages aren’t materialized in YT, and only recovery metadata is stored. For more details, see [Process functions](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md#how-it-works).

### 4. Implement a stateful process function {#implement-transform}

Use `IProcessFunction` for stateful processing as well. To work with external state, use `TSimpleExternalStateManager`:

```cpp
class TWordCountFunction
    : public IProcessFunction
{
public:
    void Init(const IRuntimeInitContextPtr& initContext) override
    {
        initContext->InitExternalStateClient(StateClient_, "/state");
    }

    void ProcessMessage(
        const TInputMessageConstPtr& message,
        const IOutputCollectorPtr& /*output*/,
        const IRuntimeContextPtr& /*context*/) override
    {
        auto state = StateClient_.GetState(message->Key);
        i64 count = state->GetColumnValue<std::optional<i64>>("count").value_or(0);
        TPayloadBuilder builder(state->Schema);
        builder.Set(count + 1, "count");
        state->Payload = builder.Finish();
    }

private:
    TMutableStateKeyClient<TSimpleExternalState> StateClient_;
};

YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction);
```

Key points:
- `TMutableStateKeyClient<TState>` is a type-safe client for external state. You set the manager’s parameters in the `Computation` spec (see below), not in your own `TParameters`/`TDynamicParameters`.
- `InitExternalStateClient(StateClient_, "/state")` binds the client to the external state manager named `"/state"`, which is declared in the `external_state_managers` spec.
- In the spec, the built-in `TProcessFunctionComputation` executes the function and provides transform-mode exactly-once state commits.

### 5. Write main.cpp {#write-main}

The `main` function ties all components together:

```cpp
#include <yt/yt/flow/library/cpp/runner/init.h>
#include <yt/yt/flow/library/cpp/runner/simple_runner_program.h>

int main(int argc, const char** argv)
{
    NYT::NFlow::Initialize(argc, argv);
    TSimpleSpecBuilder builder;
    builder.RegisterStream<TWordMessage>("words");
    return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv);
}
```

Here:
- `Initialize(argc, argv)` initializes the Flow runtime.
- `TSimpleSpecBuilder` is a builder where you register all type-safe streams. It automatically infers schemas from the registered `TYsonMessage` types.
- `RegisterStream<TWordMessage>("words")` registers the `words` stream with the `TWordMessage` message type.
- `TSimpleRunnerProgram` is a standard runner that handles starting and managing computations.

Full source code for the example:

```cpp
#include <yt/yt/flow/examples/cpp/word_count/lib/word_count_functions.h>

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


#include <yt/yt/flow/library/cpp/runner/init.h>
#include <yt/yt/flow/library/cpp/runner/simple_runner_program.h>

using namespace NYT;
using namespace NYT::NFlow;
using namespace NYT::NFlow::NExample;

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

YT_FLOW_DEFINE_YSON_MESSAGE(TWordMessage);

// Register the functions; the pipeline spec picks the hosting computation (the mode) and
// names the function via the `processing_function` parameter.
YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction, TTextReaderParameters);
YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction);

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

int main(int argc, const char** argv)
{
    NYT::NFlow::Initialize(argc, argv);
    TSimpleSpecBuilder builder;
    builder.RegisterStream<TWordMessage>("words");
    return NYT::NFlow::TSimpleRunnerProgram(std::move(builder)).Run(argc, argv);
}
```

### 6. Describe the pipeline spec {#write-spec}

The spec describes the pipeline topology in YSON format. Here’s an example for word count:

```yson
{
    "spec" = {
        "computations" = {
            "reader" = {
                "computation_class_name" = "NYT::NFlow::TProcessFunctionSourceComputation";
                "processing_function" = "NYT::NFlow::NExample::TTextReadFunction";
                "output_stream_ids" = ["words"];
                "source_streams" = {
                    "queue" = {
                        "source_class_name" = "NYT::NFlow::TQueueSource";
                        "parameters" = {
                            "queue_path" = "<cluster=cluster_name>//path/to/queue";
                            "consumer_path" = "<cluster=cluster_name>//path/to/consumer";
                        };
                    };
                };
            };
            "counter" = {
                "computation_class_name" = "NYT::NFlow::TProcessFunctionComputation";
                "processing_function" = "NYT::NFlow::NExample::TWordCountFunction";
                "input_stream_ids" = ["words"];
                "output_stream_ids" = [];
                "group_by_schema" = [
                    {"name" = "hash"; "type" = "uint64"; "expression" = "farm_hash(word)";};
                    {"name" = "word"; "type" = "string";};
                ];
                "external_state_managers" = {
                    "/state" = {
                        "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
                        "parameters" = {
                            "path" = "//path/to/state/table";
                        };
                    };
                };
            };
        };
    };
}
```

You don’t need to fill the `streams` section when you use `TYsonMessage` and `TSimpleSpecBuilder`; the stream information is inferred automatically.

For more details on the spec format, see [Spec & DynamicSpec](https://ytsaurus.tech/docs/en/flow/concepts/spec.md).

### 7. Build the project {#build}

The code fragments above explain the checked-in Word Count example; they aren't a standalone project. Build the example from the repository. Its dependencies are listed in [ya.make](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/word_count/ya.make):

```bash
ya make yt/yt/flow/examples/cpp/word_count
```

The binary is `yt/yt/flow/examples/cpp/word_count/word_count`. The simplified step 6 spec omits `min_word_length`, so the default `0` applies; the checked-in config sets it to `4`. If you move the process functions into another namespace, use their fully qualified names in the spec's `processing_function` fields.

### 8. Create objects in YT {#create-yt-objects}

Before you run the pipeline, you need to create:
- An input queue (if it doesn’t exist yet).
- A consumer for the input queue, registered on it.
- A state table (for `ExternalState`).

For a working example of creating the queue, consumer, and state table, see [Prepare the pipeline](https://ytsaurus.tech/docs/en/flow/quickstart.md#prepare) in the quick start. The C++ runner creates the pipeline and internal tables on first launch; for a separate preparation step, see [Creating a pipeline](https://ytsaurus.tech/docs/en/flow/concepts/pipeline-object.md#create).

### 9. Run and test {#run-and-test}

Save the YSON from step 6 as `<config.yson>`. Replace the queue, consumer, and external state table paths, then add `cluster_url`, the pipeline `path`, and a `vanilla` block with `enable = %true`, your scheduler `pool`, and `worker = {"count" = 1;}`, as shown in [Prepare the pipeline](https://ytsaurus.tech/docs/en/flow/quickstart.md#prepare). Launch the built binary with `--config <config.yson>` as in [Validate and launch](https://ytsaurus.tech/docs/en/flow/quickstart.md#launch). Check the state with `yt --proxy <cluster> flow get-pipeline-state <pipeline_path>`, where `<pipeline_path>` is the `path` value from your config; the CLI should print `working`.

Before you release a new version, read [Release and table change rules](https://ytsaurus.tech/docs/en/flow/devops/vanilla/releases.md#release-and-configure-basic-rules). For lifecycle commands, see [Pipeline operations](https://ytsaurus.tech/docs/en/flow/devops/vanilla/pipeline-operations.md).

## See also

- [Process functions (C++)](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md)
- [Computation modes (C++)](https://ytsaurus.tech/docs/en/flow/cpp/computation.md)
- [Working with states (C++)](https://ytsaurus.tech/docs/en/flow/cpp/state.md)
- [Watermarks](https://ytsaurus.tech/docs/en/flow/concepts/watermarks.md)
- [Timers](https://ytsaurus.tech/docs/en/flow/concepts/timers.md)
- [Spec & DynamicSpec](https://ytsaurus.tech/docs/en/flow/concepts/spec.md)
<!-- endsource: en/_includes/flow/cpp/getting-started.md -->