---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/examples/word_count.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/examples/word_count.md
  - href: https://ytsaurus.tech/docs/en/flow/cpp/examples/word_count.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/word_count.md -->
# Word Count in YTsaurus Flow (C++)

This is a simple introductory example in C++. The [pipeline](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/word_count) reads text messages from the input queue, splits them into words, and counts the occurrences of each word using an external state table.

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

Both units of user logic are written as [process functions](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md) (subclasses of `IProcessFunction`). The function itself is lightweight and testable, and its execution mode is selected by the built-in `Computation` adapter in the spec via the `processing_function` field.

## Pipeline components

### TTextReadFunction

`TTextReadFunction` is a process function (`IProcessFunction`) that is executed by `TProcessFunctionSourceComputation` (the source adapter). It reads text messages from the input queue, splits the text into words (by whitespace characters), and for each word with a length of at least `min_word_length`, it generates a `TWordMessage` object in the `words` output stream. The `min_word_length` parameter is read in `Init` via `initContext->GetParameters<TTextReaderParameters>()` from the `processing_function_parameters` block of the spec.

`TProcessFunctionSourceComputation` runs the function in Swift ordered-source mode, so output messages are not stored in YTsaurus — only the metadata required for deterministic operation is saved. For more details about computation modes, see the [Computations](https://ytsaurus.tech/docs/en/flow/concepts/computation.md) section.

### TWordCountFunction

`TWordCountFunction` is a process function (`IProcessFunction`) that is executed by `TProcessFunctionComputation` (the transform adapter). It uses `TSimpleExternalStateManager` to work with the [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state). For each input word, it:

- Gets the current counter value from the external state table.
- Increases the counter by 1.
- Writes the updated value back.

For more details about working with state, see the [State](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) section.

## Message types

`TWordMessage` is a subclass of `TYsonMessage`. It contains a single `Word` field. It is registered via the `YT_FLOW_DEFINE_YSON_MESSAGE` macro.

## Pipeline structure

The pipeline consists of two adapter computations connected by the `words` stream:

1. **source** (input queue) → **TProcessFunctionSourceComputation** (`processing_function = TTextReadFunction`) → `words` stream.
2. `words` stream → **TProcessFunctionComputation** (`processing_function = TWordCountFunction`) → state table (word → count).

In the spec for the counter, you specify `group_by_schema` with a hash of the word and the word itself — to ensure correct [partitioning](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition). The state parameters (`TSimpleExternalStateManager` and the table path) are declared in the `external_state_managers` section of the `Computation` spec.

## The main function

In `main`, you do the following:

1. `NYT::NFlow::Initialize(argc, argv)` — initialize the Flow library.
2. Register the functions via `YT_FLOW_DEFINE_PROCESS_FUNCTION(TTextReadFunction)` and `YT_FLOW_DEFINE_PROCESS_FUNCTION(TWordCountFunction)`.
3. `TSimpleSpecBuilder` — a builder for registering streams. You register the `words` stream with the `TWordMessage` message type via `RegisterStream<TWordMessage>("words")`.
4. `TSimpleRunnerProgram` — run the pipeline.

## Source code

### TTextReadFunction

```cpp
void TTextReadFunction::Init(const IRuntimeInitContextPtr& initContext)
{
    MinWordLength_ = initContext->GetParameters<TTextReaderParameters>()->MinWordLength;
}

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

### TWordCountFunction

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

void TWordCountFunction::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& /*output*/,
    const IRuntimeContextPtr& /*context*/)
{
    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();
}
```
<!-- endsource: en/_includes/flow/cpp/examples/word_count.md -->

<!-- source: en/_includes/flow/cpp/examples/word_count_also.md -->
## See also

- [Getting started (C++)](https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md)
- [Computation (C++)](https://ytsaurus.tech/docs/en/flow/cpp/computation.md)
- [Stateful processing](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md)
<!-- endsource: en/_includes/flow/cpp/examples/word_count_also.md -->