Static Table Join в YTsaurus Flow (Python)

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

Структура пайплайна

reference_reader читает статическую таблицу;
reference_loader нормализует значения и записывает их в /reference_state;
event_reader читает события из очереди;
enricher выполняет read-only lookup через joiner /reference_state и пишет успешные совпадения в enriched.

Менеджер и joiner ссылаются на одну динамическую таблицу стейта. Её ключи должны совпадать с group_by_schema компьютейшенов. См. спеку пайплайна.

Загрузка и join

Загрузчик получает стейт через ctx.external_state(...) и сохраняет нормализованное значение. Обогатитель вызывает ctx.joined_external_state(...); если normalized_name отсутствует, он не публикует результат.

class ReferenceLoader(RowFunction):
    """Trims and lowercases the reference name, storing it in external state."""

    def on_message(self, message, output, ctx):
        name = message.payload["name"]
        normalized_name = name.strip().lower()

        state = ctx.external_state("/reference_state", message)
        builder = state.to_builder()
        builder.set("normalized_name", normalized_name)
        state.set(builder.finish())

class Enricher(RowFunction):
    """Joins each event against the reference state and emits the enriched row."""

    def on_message(self, message, output, ctx):
        state = ctx.joined_external_state("/reference_state", message)
        normalized_name = state.get("normalized_name")
        if normalized_name is None:
            return

        builder = ctx.message_builder("enriched")
        builder.set("key", message.payload["key"])
        builder.set("name", normalized_name)
        output.add_message(builder.finish())

См. также

URL Downloader

External State Join