StateAccessor в YTsaurus Flow (Python)

StateAccessor — интерфейс для чтения, модификации и удаления значений стейта.
Общие сведения о stateful-обработке описаны в разделе Stateful processing.

Принцип работы

Стейт во Flow хранится в сортированных динамических таблицах.
В случае внешнего стейта эта таблица создаётся пользователем, в случае внутреннего стейта таблицы создаются и управляются Flow автоматически.

Для TTransformCompanionComputation ключевые колонки в таблице стейта совпадают с group_by_schema компьютейшена. Для внутреннего стейта TTransformOrderedSourceCompanionComputation ключом служит ключ партиции источника: group_by_schema в таком SourceComputation не поддерживается. Во всех случаях сообщения с одинаковым ключом разделяют один стейт.

Чтение и запись данных

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

Запись новых значений в таблицу стейта осуществляется транзакционно в рамках эпохи.

Четыре типа аксессоров

Python SDK предоставляет четыре вида аксессоров для работы со стейтом:

Аксессор Формат Получение Описание
YsonStateAccessor YSON (dict) ctx.state(name, msg) Сериализация Python dict в YSON
RawStateAccessor bytes ctx.raw_state(name, msg) Сырые байты без сериализации
ProtoStateAccessor Protobuf ctx.proto_state(name, msg, ProtoClass) Сериализация через Protobuf
ExternalStateAccessor Payload (строка таблицы) ctx.external_state("/name", msg) Типизированный доступ к внешней таблице

Первые три аксессора работают с внутренним стейтом (таблицы управляются Flow автоматически). ExternalStateAccessor работает с внешним стейтом (пользователь создаёт таблицу самостоятельно).

Общий API

Все внутренние аксессоры (YsonStateAccessor, RawStateAccessor, ProtoStateAccessor) предоставляют единый набор методов:

Метод Описание
get() Получить значение стейта (или None, если стейт отсутствует)
set(value) Установить значение стейта
clear() Удалить стейт для текущего ключа
get_or_default(default) Получить значение или вернуть default

Получение аксессора

Аксессор получается через RuntimeContext (ctx) внутри on_message или on_timer:

class MyFunction(RowFunction):
    def on_message(self, message, output, ctx):
        # YSON (dict)
        yson_state = ctx.state("state-name", message)

        # Raw bytes
        raw_state = ctx.raw_state("state-name", message)

        # Protobuf
        proto_state = ctx.proto_state("state-name", message, MyProtoClass)

        # External (имя обязательно начинается с "/")
        ext_state = ctx.external_state("/state-name", message)

    def on_timer(self, timer, output, ctx):
        # Аналогично, но вместо message передаётся timer
        state = ctx.state("state-name", timer)

Параметры:

  • name — строка с именем стейта, объявленным в статической спеке. Для внутренних стейтов — произвольная строка из internal_states; для внешних стейтов — ключ из external_state_managers, обязательно начинающийся с /.
  • message / timer — сообщение или таймер, для ключа которого нужно получить стейт.
  • ProtoClass (только для proto_state) — класс Protobuf-сообщения для десериализации.

Когда использовать какой тип

Ситуация Рекомендуемый аксессор
Простой словарь с несколькими полями ctx.state() (YSON)
Произвольные двоичные данные ctx.raw_state() (Raw)
Структурированные данные с фиксированной схемой ctx.proto_state() (Protobuf)
Данные, к которым нужен доступ из других систем ctx.external_state("/name", msg) (External)
Данные, требующие пользовательской таблицы ctx.external_state("/name", msg) (External)

См. также

Предыдущая
Работа со стейтами
Следующая
Internal State