External State в YTsaurus Flow (Python)

External State — механизм работы с внешним состоянием (стейтом), хранящимся во внешней динамической таблице YTsaurus. Пользователь самостоятельно создаёт таблицу для хранения стейта на том же кластере, где развёрнут пайплайн.

Для стейта в формате строк таблицы используйте ctx.external_state(...) и ctx.joined_external_state(...). Если менеджер или joiner предоставляет стейт в формате protobuf, используйте protobuf-аксессоры.

Общие сведения о stateful-обработке описаны в разделе Stateful processing.

Обзор

В Python SDK External State в формате строк таблицы представлен классом ExternalStateAccessor, который является подклассом Payload и предоставляет dict-like доступ к колонкам внешней динамической таблицы. Стейт привязан к ключу сообщения (group_by_schema). Подробнее про StateAccessor и работу со стейтом: StateAccessor.

Для read-only доступа по ключу используйте External State Joiner. Объявите его в top-level секции external_state_joiners компьютейшена, рядом с external_state_managers, и читайте через ctx.joined_external_state(...). Подробности приведены в разделе Read-only join.

Отличие от Internal State

Подробнее о Internal State см. Internal State.

Получение ExternalStateAccessor

ExternalStateAccessor получается через RuntimeContext (ctx):

# Для сообщения
state = ctx.external_state("/state-name", message)

# Для таймера
state = ctx.external_state("/state-name", timer)

Параметры:

• "/state-name" — строка с именем стейта из секции external_state_managers статической спеки. Имя обязательно начинается с / и совпадает с ключом в спеке.
• message / timer — сообщение или таймер, для ключа которого нужно получить стейт.

ExternalStateAccessor как Payload

Исходный код

ExternalStateAccessor наследует класс Payload, что позволяет читать значения колонок напрямую:

state = ctx.external_state("/shuffle-state", message)

# Dict-like доступ
value = state["count"]          # Выбросит KeyError, если колонки нет
value = state.get("count")      # Вернёт None, если колонки нет
exists = "count" in state       # Проверка наличия

Класс Payload также предоставляет:

• keys() — список имён колонок с непустыми значениями.
• to_dict() — конвертация в обычный Python dict.

Основные операции

Чтение стейта

state = ctx.external_state("/join-state", message)

# Чтение значения колонки
hit_payload = state.get("hit_payload", str)
show_time = state.get("show_time")

# Dict-like доступ
try:
    value = state["hit_payload"]
except KeyError:
    value = None

Запись стейта через PayloadBuilder

Для модификации стейта используется паттерн to_builder() / set() / finish():

state = ctx.external_state("/join-state", message)

# Получить builder с текущими значениями
builder = state.to_builder()

# Обновить нужные поля
builder.set("hit_payload", "some_value")
builder.set("show_time", 1234567890)

# Сохранить обновлённый стейт
state.set(builder.finish())

Метод to_builder() возвращает PayloadBuilder, предзаполненный текущими значениями стейта. Метод builder.set(column, value) возвращает сам builder (поддерживает цепочку вызовов). Метод builder.finish() создаёт новый Payload и сбрасывает builder.

Очистка стейта

state = ctx.external_state("/join-state", timer)

# Удаление строки из таблицы
state.clear()

Внешний стейт в формате protobuf

ctx.proto_external_state(name, message_or_timer, proto_class) возвращает ProtoExternalStateAccessor. ctx.joined_proto_external_state(name, message_or_timer, proto_class) возвращает тот же аксессор в режиме только для чтения. Передайте класс protobuf-сообщения в proto_class и сообщение или таймер с ключом. Имя должно соответствовать правилам именования внешнего стейта и записи в external_state_managers или, для присоединённого стейта, в external_state_joiners.

Настроенный менеджер или joiner должен предоставлять protobuf-стейт: его дескриптор для компаньона должен указывать формат proto и proto_type, равный proto_class.DESCRIPTOR.full_name. Выбор аксессора не меняет формат менеджера. TSimpleExternalStateManager и TSimpleExternalStateJoiner предоставляют стейт в формате строк таблицы и не подходят для protobuf-аксессоров. Конкретная реализация профильного стейта относится к внутреннему расширению; приведённая ниже конфигурация строкового стейта не включает внешний protobuf-стейт.

Аксессор поддерживает следующие операции:

Изменения возвращённого сообщения сохраняются только после set(). Это отличается от внутреннего стейта с живыми значениями. Запись пустого protobuf-сообщения сохраняет его как значение, а clear() сбрасывает стейт. Для одного ключа действует последний явный вызов set() или clear(). Присоединённый стейт доступен только для чтения: set() и clear() выбрасывают ReadOnlyExternalStateError, а изменение возвращённого снапшота не влияет на стейт.

Неизвестное имя стейта, отсутствие ключа, неверный формат или несовпадение protobuf-типа приводят к ValueError. Если передать экземпляр сообщения вместо класса или значение неверного типа в set(), возникнет TypeError. Повреждённые сохранённые сообщения, сообщения без обязательных полей и ошибки сериализации приводят к ValueError. Ошибка сериализации сохраняет предыдущее значение стейта.

См. реализацию аксессора и тесты внешнего protobuf-стейта.

Конфигурация строкового стейта в статической спеке

Для External State в формате строк таблицы объявите external state manager в секции external_state_managers компьютейшена в статической спеке:

"computations" = {
    "reducer" = {
        "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
        "group_by_schema" = [
            {"name" = "hash"; "expression" = "farm_hash(key)"; "type" = "uint64"};
            {"name" = "key"; "type" = "string"};
        ];
        "input_stream_ids" = ["input"];
        "output_stream_ids" = ["output"];
        "external_state_managers" = {
            "/shuffle-state" = {
                "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
                "parameters" = {
                    "path" = "//path/to/state/table";
                };
            };
        };
        "parameters" = {};
    };
};

Ключевые поля:

• external_state_managers — секция верхнего уровня внутри Computation с описанием внешних state-менеджеров (раньше parameters/external_states).
• Ключ внутри external_state_managers (например, "/shuffle-state") — имя стейта, используемое в Python-коде при вызове ctx.external_state("/shuffle-state", message). Имя обязательно начинается с /.
• external_state_manager_class_name — имя зарегистрированного класса external state manager. Для стейта в формате строк таблицы используйте "NYT::NFlow::TSimpleExternalStateManager". Подробнее см. в C++ документации.
• parameters.path — путь к динамической таблице YTsaurus, в которой хранится стейт.

Read-only join

External State Joiner загружает строку, ключ которой совпадает с ключом текущего сообщения или таймера. Объявите NYT::NFlow::TSimpleExternalStateJoiner в external_state_joiners; имя стейта должно начинаться с / и совпадать с аргументом ctx.joined_external_state. См. спеки запускаемых примеров External State Join и Static Table Join.

Аксессор поддерживает те же операции чтения, что и ExternalStateAccessor, но set() и clear() выбрасывают ReadOnlyExternalStateError. Если строка не найдена, возвращается пустой payload со схемой присоединённой таблицы, поэтому get("column") вернёт None; обработчики в примерах не создают для такого сообщения выходную запись. Неизвестное имя joiner-а приводит к ValueError.

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())

Joiner использует отдельный TTL-кэш в каждой джобе. По умолчанию кэш отключён (ttl = 0), и в следующей эпохе ключи загружаются заново. Чтобы повторно использовать строки до истечения TTL, настройте dynamic_spec.computations.<computation>.external_state_joiners.<name>.parameters.cache. Присоединённые строки загружаются как неизменённый стейт и не отправляются воркеру как обновления.

См. реализацию joiner-а, реализацию TTL-кэша, преобразование присоединённого стейта в Python, хранилище стейта и read-only-аксессор.

Создание таблицы для стейта

Таблица для External State должна быть создана заранее. Схема ключевых колонок таблицы должна совпадать с group_by_schema компьютейшена.

Полный пример — EventReducer из Shuffle

Пример из shuffle:

class EventReducer(RowFunction):
    """Counts events per key using external state."""

    def on_message(self, message, output, ctx):
        state = ctx.external_state("/shuffle-state", message)
        builder = state.to_builder()
        builder.set("count", (state.get("count") or 0) + 1)
        state.set(builder.finish())

Полный исходный код

Паттерн работы:

1. Получить текущее состояние через ctx.external_state(...).
2. Создать builder через state.to_builder().
3. Обновить нужные поля через builder.set(...).
4. Сохранить через state.set(builder.finish()).

См. также

• StateAccessor (Python)
• Internal State (Python)
• Работа со стейтами (Python) — краткий обзор
• Stateful processing (концепция)
• External State (Java)
• Пример Shuffle
• Пример Static Table Join
• Пример External State Join