--- metadata: - name: generator content: Diplodoc Platform v5.63.3 alternate: - https://ytsaurus.tech/docs/en/flow/python/internal-state.md - https://ytsaurus.tech/docs/ru/flow/python/internal-state.md - href: https://ytsaurus.tech/docs/ru/flow/python/internal-state.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/ru/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Internal State в YTsaurus Flow (Python) Internal State — механизм работы с внутренним [состоянием (стейтом)](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#state), хранящимся во внутренних таблицах Flow. В отличие от [External State](https://ytsaurus.tech/docs/ru/flow/python/external-state.md), пользователю не нужно самостоятельно создавать таблицы — Flow управляет ими автоматически. Подробнее про `StateAccessor` и общие принципы работы со стейтом: [StateAccessor](https://ytsaurus.tech/docs/ru/flow/python/state-accessor.md). Общие сведения о stateful-обработке описаны в разделе [Stateful processing](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md). ## Обзор Python SDK предоставляет три вида аксессоров для работы с Internal State, различающихся форматом сериализации. Для работы с внешним стейтом используется отдельный аксессор — [ExternalStateAccessor](https://ytsaurus.tech/docs/ru/flow/python/external-state.md). | Аксессор | Формат | Описание | |----------|--------|----------| | [YsonStateAccessor](#yson-state-accessor) | YSON (dict) | Сериализация Python dict через YSON | | [RawStateAccessor](#raw-state-accessor) | `bytes` | Без сериализации (сырые байты) | | [ProtoStateAccessor](#proto-state-accessor) | Protobuf | Сериализация через Protobuf | Все аксессоры предоставляют единый набор методов: `get()`, `set(value)`, `clear()`, `get_or_default(default)`, `read_only()`. ## Изменение значения на месте {#in-place} Значение, которое возвращают `get()` и `get_or_default()`, живое: для каждого ключа оно декодируется один раз за батч, все аксессоры этого ключа возвращают один и тот же объект, а изменения, сделанные в нём, записываются в стейт по окончании батча без вызова `set()`. Если значение не изменилось, запись не выполняется. Дефолт из `get_or_default()` привязывается к ключу, но не записывается: его можно сразу изменять, а значением стейта он становится только после изменения, поэтому нетронутый дефолт не создаёт строку в таблице стейтов: ```python 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`: ```python count = ctx.state("word-state", message).read_only().get_or_default({"count": 0})["count"] ``` Значение общее, а не копия: если раньше в этом же батче ключ прочитал пишущий аксессор, изменение через read-only-представление всё равно попадёт в стейт. ## YsonStateAccessor {#yson-state-accessor} [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `YsonStateAccessor` использует YSON-сериализацию. Стейт хранится как Python dict, автоматически сериализуемый в YSON и обратно. ### Получение аксессора ```python # Для сообщения state = ctx.state("state-name", message) # Для таймера state = ctx.state("state-name", timer) ``` ### Методы | Метод | Возвращаемый тип | Описание | |-------|-----------------|----------| | `get()` | `dict` или `None` | Десериализовать и вернуть текущее значение | | `set(value)` | — | Сериализовать и сохранить значение (dict или bytes) | | `clear()` | — | Удалить стейт для текущего ключа | | `get_or_default(default)` | `dict` | Вернуть текущее значение или `default` (записывается только после изменения) | ### Пример из WordCount ```python 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) ``` [Полный исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/word_count/word_count_mapper.py) Здесь стейт привязан к ключу [сообщения](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#message) (определяемому через `group_by_schema` в [спеке](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#spec-and-dynamic-spec)). Для каждого уникального слова хранится свой независимый счётчик. ## RawStateAccessor {#raw-state-accessor} [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `RawStateAccessor` работает с сырыми байтами без сериализации/десериализации. ### Получение аксессора ```python # Для сообщения state = ctx.raw_state("state-name", message) # Для таймера state = ctx.raw_state("state-name", timer) ``` ### Методы | Метод | Возвращаемый тип | Описание | |-------|-----------------|----------| | `get()` | `bytes` или `None` | Получить сырые байты | | `set(value: bytes)` | — | Сохранить сырые байты | | `clear()` | — | Удалить стейт для текущего ключа | | `get_or_default(default: bytes)` | `bytes` | Вернуть текущее значение или привязать и вернуть `default` (в YTsaurus попадает только через `set()`) | ### Пример использования ```python 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 {#proto-state-accessor} [Исходный код](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `ProtoStateAccessor` использует Protobuf-сериализацию. Стейт десериализуется в экземпляр указанного Protobuf-класса. ### Получение аксессора ```python # Для сообщения state = ctx.proto_state("state-name", message, TJoinState) # Для таймера state = ctx.proto_state("state-name", timer, TJoinState) ``` Третий аргумент — класс Protobuf-сообщения, используемый для десериализации. ### Методы | Метод | Возвращаемый тип | Описание | |-------|-----------------|----------| | `get()` | Proto-объект или `None` | Десериализовать и вернуть значение | | `set(value)` | — | Сериализовать и сохранить Proto-объект | | `clear()` | — | Удалить стейт для текущего ключа | | `get_or_default(default=None)` | Proto-объект | Вернуть значение, `default` либо пустой экземпляр Proto-класса (записывается только после изменения) | {% note info %} Метод `get_or_default()` без аргументов возвращает пустой экземпляр Proto-класса (эквивалент `ProtoClass()`). Это удобно для инициализации стейта при первом обращении; сам стейт появится, когда вы измените это сообщение. {% endnote %} ## Конфигурация в статической спеке {#static-spec} Internal State не требует создания внешних таблиц. Стейты автоматически хранятся во внутренних таблицах Flow. Имена внутренних стейтов должны быть объявлены в секции `internal_states` параметров [компьютейшена](https://ytsaurus.tech/docs/ru/flow/concepts/glossary.md#stream-and-computation) в статической спеке: ```yson { "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" = "//path/to/queue"; "consumer_path" = "//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`. {% note warning %} Если имя стейта не объявлено в `internal_states`, при обращении к нему через `ctx.state(...)` будет выброшено исключение `ValueError`. {% endnote %} ## См. также - [StateAccessor (Python)](https://ytsaurus.tech/docs/ru/flow/python/state-accessor.md) - [External State (Python)](https://ytsaurus.tech/docs/ru/flow/python/external-state.md) - [Работа со стейтами (Python)](https://ytsaurus.tech/docs/ru/flow/python/state.md) — краткий обзор - [Stateful processing (концепция)](https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md) - [Internal State (Java)](https://ytsaurus.tech/docs/ru/flow/java/internal-state.md) - [Пример WordCount](https://ytsaurus.tech/docs/ru/flow/python/examples/wordcount.md)