External State Join в YTsaurus Flow (Python)

Пайплайн обогащает события из очереди данными из заранее созданной динамической таблицы через read-only external state joiner.

Исходники

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

event_reader публикует стрим event. Компаньонский компьютейшен lookup_join группирует сообщения по key, читает /reference через external_state_joiners и передаёт успешные совпадения в enriched. Queue-sink сохраняет выходные сообщения. path joiner-а может указывать на Cypress-ссылку, чтобы переключать справочник без перезапуска пайплайна.

См. спеку пайплайна.

Чтение присоединённого стейта

ctx.joined_external_state("/reference", message) выбирает строку по ключу сообщения. Возвращаемый аксессор доступен только для чтения. Если строка не найдена, он содержит пустой payload: get("name") возвращает None, и пример не публикует результат.

class LookupJoin(RowFunction):
    """Looks up each event's name in the reference state and emits the joined row."""

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

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

См. также

Static Table Join

Быстрый старт