--- 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/en/flow/python/internal-state.md type: text/markdown title: Markdown version - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 rel: describedby --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65 # Internal State in YTsaurus Flow (Python) Internal State is a mechanism for working with internal [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state) stored in Flow’s internal tables. Unlike [External State](https://ytsaurus.tech/docs/en/flow/python/external-state.md), you don’t need to create tables manually—Flow manages them automatically. Learn more about `StateAccessor` and general principles for working with state: [StateAccessor](https://ytsaurus.tech/docs/en/flow/python/state-accessor.md). General information about stateful processing is in the [Stateful processing](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) section. ## Overview The Python SDK provides three types of accessors for working with Internal State, which differ by serialization format. To work with external state, use a separate accessor: [ExternalStateAccessor](https://ytsaurus.tech/docs/en/flow/python/external-state.md). | Accessor | Format | Description | |----------|--------|----------| | [YsonStateAccessor](#yson-state-accessor) | YSON (dict) | Serializes a Python dict using YSON | | [RawStateAccessor](#raw-state-accessor) | `bytes` | No serialization (raw bytes) | | [ProtoStateAccessor](#proto-state-accessor) | Protobuf | Serializes using Protobuf | All accessors provide the same set of methods: `get()`, `set(value)`, `clear()`, `get_or_default(default)`, `read_only()`. ## Changing the value in place {#in-place} The value returned by `get()` and `get_or_default()` is live: it is decoded once per key and batch, every accessor for that key returns the same object, and the changes made to it are written to the state at the end of the batch without a `set()` call. Nothing is written when the value did not change. The default from `get_or_default()` is attached to the key but not written: it can be changed right away, and it becomes the state value only once it is changed, so an untouched default creates no row in the state table: ```python ctx.state("word-state", message).get_or_default({"count": 0})["count"] += 1 ``` `set()` still replaces the whole value, and `clear()` removes the state. A `None` default is not a value and is not attached. A value that encodes to no bytes — an all-default protobuf message, for instance — is not a value either: setting it removes the state. A `RawStateAccessor` default is attached like any other, and a later `get()` returns it. But a `bytes` object is immutable, so an attached default cannot be changed in place, and a non-empty raw value reaches YTsaurus only through `set()`. To detect the changes, a value that was read is re-encoded at the end of the batch. When the computation only reads a state, use `read_only()`: that accessor returns the same object but does not track it, its `get_or_default()` does not create the state, and `set()` and `clear()` raise `ReadOnlyStateError`: ```python count = ctx.state("word-state", message).read_only().get_or_default({"count": 0})["count"] ``` The value is shared, not a copy: if a writable accessor read the same key earlier in the batch, a change made through the read-only view still reaches the state. ## YsonStateAccessor {#yson-state-accessor} [Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `YsonStateAccessor` uses YSON serialization. The state is stored as a Python dict, which is automatically serialized to YSON and deserialized back. ### Get the accessor ```python # For a message state = ctx.state("state-name", message) # For a timer state = ctx.state("state-name", timer) ``` ### Methods | Method | Return type | Description | |-------|-----------------|----------| | `get()` | `dict` or `None` | Deserialize and return the current value | | `set(value)` | — | Serialize and save the value (dict or bytes) | | `clear()` | — | Delete the state for the current key | | `get_or_default(default)` | `dict` | Return the current value, or `default` (written only once you change it) | ### Example from 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) ``` [Full source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/python/word_count/word_count_mapper.py) Here, the state is tied to the [message](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#message) key (defined via `group_by_schema` in the [spec](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#spec-and-dynamic-spec)). Each unique word has its own independent counter. ## RawStateAccessor {#raw-state-accessor} [Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `RawStateAccessor` works with raw bytes without serialization or deserialization. ### Get the accessor ```python # For a message state = ctx.raw_state("state-name", message) # For a timer state = ctx.raw_state("state-name", timer) ``` ### Methods | Method | Return type | Description | |-------|-----------------|----------| | `get()` | `bytes` or `None` | Get the raw bytes | | `set(value: bytes)` | — | Save the raw bytes | | `clear()` | — | Delete the state for the current key | | `get_or_default(default: bytes)` | `bytes` | Return the current value, or attach and return `default` (reaches YTsaurus only through `set()`) | ### Usage example ```python state = ctx.raw_state("raw-state", message) data = state.get() if data is not None: # Process raw data... pass # Write raw data state.set(b"\x01\x02\x03") # Clear state.clear() ``` ## ProtoStateAccessor {#proto-state-accessor} [Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/python/companion/context.py) `ProtoStateAccessor` uses Protobuf serialization. The state is deserialized into an instance of the specified Protobuf class. ### Get the accessor ```python # For a message state = ctx.proto_state("state-name", message, TJoinState) # For a timer state = ctx.proto_state("state-name", timer, TJoinState) ``` The third argument is the Protobuf message class used for deserialization. ### Methods | Method | Return type | Description | |-------|-----------------|----------| | `get()` | Proto object or `None` | Deserialize and return the value | | `set(value)` | — | Serialize and save the Proto object | | `clear()` | — | Delete the state for the current key | | `get_or_default(default=None)` | Proto object | Return the value, `default`, or an empty instance of the Proto class (written only once you change it) | {% note info %} The `get_or_default()` method with no arguments returns an empty instance of the Proto class (equivalent to `ProtoClass()`). This is convenient for initializing the state on the first access; the state itself appears once you change that message. {% endnote %} ## Configuration in the static spec {#static-spec} Internal State doesn’t require you to create external tables. States are automatically stored in Flow’s internal tables. You must declare the names of internal states in the `internal_states` section of the [computation](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation) parameters in the static spec: ```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" = { }; }; }; }; } ``` The state name in your code (the first argument of `ctx.state(...)`, `ctx.raw_state(...)`, or `ctx.proto_state(...)`) must match the name declared in `internal_states`. {% note warning %} If the state name isn’t declared in `internal_states`, calling it via `ctx.state(...)` will raise a `ValueError` exception. {% endnote %} ## See also - [StateAccessor (Python)](https://ytsaurus.tech/docs/en/flow/python/state-accessor.md) - [External State (Python)](https://ytsaurus.tech/docs/en/flow/python/external-state.md) - [Working with states (Python)](https://ytsaurus.tech/docs/en/flow/python/state.md) — brief overview - [Stateful processing (concept)](https://ytsaurus.tech/docs/en/flow/concepts/stateful.md) - [Internal State (Java)](https://ytsaurus.tech/docs/en/flow/java/internal-state.md) - [WordCount example](https://ytsaurus.tech/docs/en/flow/python/examples/wordcount.md)