Internal State в YTsaurus Flow (Python)

Internal State — механизм работы с внутренним состоянием (стейтом), хранящимся во внутренних таблицах Flow. В отличие от External State, пользователю не нужно самостоятельно создавать таблицы — Flow управляет ими автоматически.

Подробнее про StateAccessor и общие принципы работы со стейтом: StateAccessor.

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

Обзор

Python SDK предоставляет три вида аксессоров для работы с Internal State, различающихся форматом сериализации. У внешнего стейта отдельные строковые и protobuf-аксессоры, включая варианты для чтения присоединённого стейта.

Все аксессоры предоставляют единый набор методов: get(), set(value), clear(), get_or_default(default), read_only().

Изменение значения на месте

Описанное в этом разделе поведение живых значений относится к внутренним аксессорам. Аксессоры внешнего protobuf-стейта возвращают независимые снапшоты; для сохранения изменений нужен явный вызов set().

Значение, которое возвращают get() и get_or_default(), живое: для каждого ключа оно декодируется один раз за батч, все аксессоры этого ключа возвращают один и тот же объект, а изменения, сделанные в нём, записываются в стейт по окончании батча без вызова set(). Если значение не изменилось, запись не выполняется. Дефолт из get_or_default() привязывается к ключу, но не записывается: его можно сразу изменять, а значением стейта он становится только после изменения, поэтому нетронутый дефолт не создаёт строку в таблице стейтов:

ctx.state("word-state", message).get_or_default({"count": 0})["count"] += 1

set() по-прежнему заменяет значение целиком, clear() удаляет стейт. Дефолт None значением не является и не привязывается. Значение, которое кодируется в пустые байты — например, protobuf-сообщение со всеми полями по умолчанию, — тоже значением не является: его запись удаляет стейт. Дефолт get_or_default() у RawStateAccessor привязывается так же, как у остальных аксессоров, и его вернёт последующий get(). Но объект bytes неизменяем, изменить привязанный дефолт на месте нельзя, поэтому непустое значение сырого стейта попадает в YTsaurus только через set().

Чтобы отследить изменения, прочитанное значение перекодируется по окончании батча. Если вычисление только читает стейт, используйте read_only(): такой аксессор возвращает тот же объект, но не отслеживает его, get_or_default() не создаёт стейт, а set() и clear() бросают ReadOnlyStateError:

count = ctx.state("word-state", message).read_only().get_or_default({"count": 0})["count"]

Значение общее, а не копия: если раньше в этом же батче ключ прочитал пишущий аксессор, изменение через read-only-представление всё равно попадёт в стейт.

YsonStateAccessor

Исходный код

YsonStateAccessor использует YSON-сериализацию. Стейт хранится как Python dict, автоматически сериализуемый в YSON и обратно.

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

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

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

Методы

Пример из WordCount

class WordCountMapper(RowFunction):
    """Processes Word messages and maintains word count state."""

    def on_message(self, message, output, ctx):
        state = ctx.state("word-state", message)
        data = state.get_or_default({"word": message.payload["word"], "count": 0})
        data["count"] += 1
        state.set(data)

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

Здесь стейт привязан к ключу сообщения (определяемому через group_by_schema в спеке). Для каждого уникального слова хранится свой независимый счётчик.

RawStateAccessor

Исходный код

RawStateAccessor работает с сырыми байтами без сериализации/десериализации.

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

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

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

Методы

Пример использования

state = ctx.raw_state("raw-state", message)

data = state.get()
if data is not None:
    # Обработка сырых данных...
    pass

# Запись сырых данных
state.set(b"\x01\x02\x03")

# Очистка
state.clear()

ProtoStateAccessor

Исходный код

ProtoStateAccessor использует Protobuf-сериализацию. Стейт десериализуется в экземпляр указанного Protobuf-класса.

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

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

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

Третий аргумент — класс Protobuf-сообщения, используемый для десериализации.

Методы

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

Internal State не требует создания внешних таблиц. Стейты автоматически хранятся во внутренних таблицах Flow.

Имена внутренних стейтов должны быть объявлены в секции internal_states параметров компьютейшена в статической спеке:

{
    "spec" = {
        "computations" = {
            "reader" = {
                "computation_class_name" = "NYT::NFlow::TSwiftPassthroughOrderedSourceComputation";
                "output_stream_ids" = ["words"];
                "source_streams" = {
                    "queue" = {
                        "source_class_name" = "NYT::NFlow::TQueueSource";
                        "parameters" = {
                            "queue_path" = "<cluster=cluster_name>//path/to/queue";
                            "consumer_path" = "<cluster=cluster_name>//path/to/consumer";
                            "finite" = false;
                        };
                    };
                };
                "parameters" = {};
            };
            "mapper" = {
                "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
                "group_by_schema" = [
                    {"name" = "hash"; "expression" = "farm_hash(word)"; "type" = "uint64"; required = %true;};
                    {"name" = "word"; "type" = "string";};
                ];
                "input_stream_ids" = ["words"];
                "output_stream_ids" = [];
                "required_resource_ids" = {
                    "CompanionManager" = {
                        "worker" = true;
                        "controller" = false;
                    };
                };
                "parameters" = {
                    "internal_states" = ["word-state"];
                };
            };
        };
        "streams" = {
            "words" = {
                "schema" = [
                    {name = "word"; type = "string";};
                ];
            };
        };
        "resources" = {
            "CompanionManager" = {
                "resource_class_name" = "NYT::NFlow::NCompanion::TCompanionManager";
                "parameters" = {
                };
                "dependencies" = {};
            };
        };
    };
    "dynamic_spec" = {
        "computations" = {
            "reader" = {
                "parameters" = {
                };
            };
            "mapper" = {
                "parameters" = {
                };
            };
        };
    };
}

Имя стейта в коде (первый аргумент ctx.state(...), ctx.raw_state(...) или ctx.proto_state(...)) должно совпадать с именем, объявленным в internal_states.

См. также

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