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

<!-- source: ru/_includes/flow/java/examples/static_table_join.md -->
# Static Table Join в YTsaurus Flow (Java и Kotlin)

Пайплайн загружает статический справочник во внешний стейт и обогащает события из очереди по общему ключу. Java и Kotlin используют один SDK и одинаковую топологию пайплайна.

[Исходники Java](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/static_table_join)

[Исходники Kotlin](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/static_table_join)

## Структура пайплайна {#structure}

- `reference_reader` читает статический справочник;
- `reference_loader` нормализует значения и записывает их в `/reference_state` через `external_state_managers`;
- `event_reader` читает события из очереди;
- `enricher` читает `/reference_state` через `external_state_joiners` и публикует успешные совпадения в `enriched`.

Менеджер и joiner ссылаются на одну динамическую таблицу стейта. Её ключи должны совпадать с `group_by_schema` компьютейшенов. См. [спеку Java](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/java/static_table_join/static_table_join/src/main/resources/pipeline.yson) или [спеку Kotlin](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/kotlin/static_table_join/static_table_join/src/main/resources/pipeline.yson).

## Загрузка справочника {#load}

{% list tabs group=lang %}

- Java

  ```java
  @Override
  public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
      String name = message.get("name", String.class);
      String normalizedName = name.strip().toLowerCase();
  
      ExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
      stateAccessor.set(stateAccessor.getOrDefault().toBuilder()
              .set("normalized_name", normalizedName)
              .finish());
  }
  ```

- Kotlin

  ```kotlin
  override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
      val normalizedName = message.get("name", String::class.java)?.trim()?.lowercase() ?: return
  
      val stateAccessor = ctx.getState(REFERENCE_STATE, message)
      stateAccessor.set(
          stateAccessor.orDefault.toBuilder()
              .set("normalized_name", normalizedName)
              .finish()
      )
  }
  ```

{% endlist %}

## Обогащение событий {#enrich}

Read-only дескриптор выбирает строку по ключу текущего сообщения. Если строка или `normalized_name` не найдены, пример не публикует результат.

{% list tabs group=lang %}

- Java

  ```java
  @Override
  public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
      ReadOnlyExternalStateAccessor stateAccessor = ctx.getState(REFERENCE_STATE, message);
      String normalizedName = stateAccessor.getOrDefault().get("normalized_name", String.class);
      if (normalizedName == null) {
          return;
      }
  
      output.addMessage(ctx.createMessageBuilder("enriched")
              .set("key", message.get("key", Long.class))
              .set("name", normalizedName)
              .finish());
  }
  ```

- Kotlin

  ```kotlin
  override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
      val stateAccessor = ctx.getState(REFERENCE_STATE, message)
      val normalizedName = stateAccessor.orDefault.get("normalized_name", String::class.java) ?: return
  
      output.addMessage(
          ctx.createMessageBuilder("enriched")
              .set("key", message.get("key", Long::class.java))
              .set("name", normalizedName)
              .finish()
      )
  }
  ```

{% endlist %}

Оба компьютейшена находятся по `@FlowComputation`; схемы трёх стримов, которые используют нативные источники и компаньонские функции, объявлены в спеке пайплайна.

## См. также

- [External State (Java)](https://ytsaurus.tech/docs/ru/flow/java/external-state.md)
- [Пример External State Join](https://ytsaurus.tech/docs/ru/flow/java/examples/external_state_join.md)
<!-- endsource: ru/_includes/flow/java/examples/static_table_join.md -->