Internal State in YTsaurus Flow (Python)
Internal State is a mechanism for working with internal state stored in Flow’s internal tables. Unlike External State, you don’t need to create tables manually—Flow manages them automatically.
Learn more about StateAccessor and general principles for working with state: StateAccessor.
General information about stateful processing is in the Stateful processing 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.
| Accessor | Format | Description |
|---|---|---|
| YsonStateAccessor | YSON (dict) | Serializes a Python dict using YSON |
| RawStateAccessor | bytes |
No serialization (raw bytes) |
| ProtoStateAccessor | 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
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:
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:
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
YsonStateAccessor uses YSON serialization. The state is stored as a Python dict, which is automatically serialized to YSON and deserialized back.
Get the accessor
# 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
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)
Here, the state is tied to the message key (defined via group_by_schema in the spec). Each unique word has its own independent counter.
RawStateAccessor
RawStateAccessor works with raw bytes without serialization or deserialization.
Get the accessor
# 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
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
ProtoStateAccessor uses Protobuf serialization. The state is deserialized into an instance of the specified Protobuf class.
Get the accessor
# 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
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.
Configuration in the 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 parameters in the static spec:
{
"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" = {
};
};
};
};
}
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.
Warning
If the state name isn’t declared in internal_states, calling it via ctx.state(...) will raise a ValueError exception.