---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/java/examples/wordcount.md
  - https://ytsaurus.tech/docs/ru/flow/java/examples/wordcount.md
  - href: https://ytsaurus.tech/docs/ru/flow/java/examples/wordcount.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: ru/_includes/flow/java/examples/wordcount.md -->
# Word Count в YTsaurus Flow (Java)

[Пайплайн](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#pipeline) читает [поток](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) слов и подсчитывает количество вхождений каждого слова с использованием YSON-стейта. Пример демонстрирует конфигурацию [компаньона](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#companion) через Spring Boot.

[Исходный код (Java)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/word_count)
[Исходный код (Kotlin)](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/word_count)

## Компоненты

### WordCountApplication

Единственная точка входа: класс с `@SpringBootApplication` запускает пайплайн, когда `YT_FLOW_MODE` не задана, и обслуживает его как компаньон, когда воркер выставляет `YT_FLOW_MODE=Worker`:

{% list tabs group=lang %}

- Java

  ```java
  @SpringBootApplication
  public class WordCountApplication {
      public static void main(String[] args) {
          new SpringApplicationBuilder(WordCountApplication.class)
                  .run(args);
      }
  }
  ```

- Kotlin

  ```kotlin
  @SpringBootApplication
  open class WordCountApplication {
      companion object {
          @JvmStatic
          fun main(args: Array<String>) {
              SpringApplicationBuilder(WordCountApplication::class.java)
                  .run(*args)
          }
      }
  }
  ```

{% endlist %}

gRPC-сервер поднимается автоматически через Spring Boot auto-config.

### Регистрация стримов

Типизированные стримы объявляются декларативно: POJO-класс сообщения помечается аннотацией `@FlowMessage` со списком идентификаторов стримов (`streamIds`), которые он обслуживает. Класс уже помечен JPA-аннотацией `@Entity`, из которой выводится схема. Spring Boot находит такие классы сканированием пакетов приложения и регистрирует стримы автоматически. Компьютейшен `mapper` регистрируется аннотацией `@FlowComputation` на классе `WordCountMapper` (см. ниже):

{% list tabs group=lang %}

- Java

  ```java
  @Entity
  @FlowMessage(streamIds = {"words"})
  public class Word {
      private String word;
  
      public Word() {
      }
  
      public Word(String word) {
          this.word = word;
      }
  
      public String getWord() {
          return word;
      }
  
      public void setWord(String word) {
          this.word = word;
      }
  }
  ```

- Kotlin

  ```kotlin
  @Entity
  @FlowMessage(streamIds = ["words"])
  class Word {
      var word: String? = null
  
      constructor()
  
      constructor(word: String?) {
          this.word = word
      }
  }
  ```

{% endlist %}

- `reader` — SourceComputation без процессной функции. Чтение и парсинг выполняются на стороне C++ [worker](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#worker)-а.
- `mapper` — Computation, реализованный классом `WordCountMapper` с аннотацией `@FlowComputation(id = "mapper")`.
- `@FlowMessage(streamIds = {"words"})` на классе `Word` — регистрирует типизированный стрим `"words"`, что позволяет получать сообщения как объекты `Word`.

### WordCountMapper

Процессная функция, реализующая подсчет слов с использованием [YsonStateAccessor](https://ytsaurus.tech/docs/ru/flow/java/state.md#yson-state):

{% list tabs group=lang %}

- Java

  ```java
      @Override
      public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
          Word input = message.getPayload();
          StateAccessor<WordCountState> stateAccessor = ctx.getState(WORD_STATE, message);
          var state = stateAccessor.getOrDefault(new WordCountState(input.getWord(), 0));
          state.setCount(state.getCount() + 1);
      }
  ```

- Kotlin

  ```kotlin
      override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
          val input: Word = message.getPayload()
          val stateAccessor = ctx.getState(WORD_STATE, message)
          val state = stateAccessor.getOrDefault(WordCountState(input.word, 0))
          state.count += 1
      }
  ```

{% endlist %}

Аннотация `@FlowComputation(id = "mapper")` регистрирует класс как компьютейшен и делает его Spring-бином (она мета-аннотирована `@Component`).

## Ключевые паттерны

- **Spring Boot auto-config** — не нужно вручную создавать `PipelineContext` и `GrpcServerExecution`.
- **@FlowComputation** — процессная функция одновременно становится Spring-бином и компьютейшеном; можно использовать инъекцию зависимостей.
- **@FlowMessage** — регистрация типизированных стримов через классы сообщений.
- **FlowStreams.typed** — типизированный доступ к сообщениям через Java-объекты.

## Запуск

Пайплайн запускается одним классом `WordCountApplication`:

```bash
./run.sh tech.ytsaurus.flow.examples.wordcount.WordCountApplication --config pipeline.yson --flow-bin flow_server
```

Он же служит компаньоном: воркер запускает его с `YT_FLOW_MODE=Worker`, и в этом режиме стартер поднимает gRPC-сервер вместо запуска пайплайна.
<!-- endsource: ru/_includes/flow/java/examples/wordcount.md -->

<!-- source: ru/_includes/flow/java/examples/wordcount_also.md -->
## См. также

- [Быстрый старт](https://ytsaurus.tech/docs/ru/flow/java/getting-started.md)
- [Computation](https://ytsaurus.tech/docs/ru/flow/java/computation.md)
- [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md)
<!-- endsource: ru/_includes/flow/java/examples/wordcount_also.md -->