External State в YTsaurus Flow (Python)

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

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

Обзор

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

Отличие от 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()

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

Для использования 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, в которой хранится стейт.

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

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

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

Пример из shuffle:

from yt.yt.flow.library.python.companion.computation import RowFunction


class EventReducer(RowFunction):
    """Подсчитывает количество событий для каждого ключа через 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