External State in YTsaurus Flow (Python)

Use External State to work with external state stored in an external dynamic table in YTsaurus. You create the table to store the state yourself, on the same cluster where the pipeline is deployed.

For general information about stateful processing, see the Stateful processing section.

Overview

In the Python SDK, External State is represented by the ExternalStateAccessor class. This class is a subclass of Payload and gives you dict-like access to the columns of the external dynamic table. The state is bound to the message key (group_by_schema). For more details about StateAccessor and working with state, see StateAccessor.

Difference from Internal State

For more details about Internal State, see Internal State.

Getting ExternalStateAccessor

You get ExternalStateAccessor through RuntimeContext (ctx):

# For a message
state = ctx.external_state("/state-name", message)

# For a timer
state = ctx.external_state("/state-name", timer)

Parameters:

• "/state-name" — a string with the state name from the external_state_managers section of the static spec. The name must start with / and match the key in the spec.
• message / timer — the message or timer for which you need to get the state for the key.

ExternalStateAccessor as Payload

Source code

ExternalStateAccessor inherits from the Payload class, which lets you read column values directly:

state = ctx.external_state("/shuffle-state", message)

# Dict-like access
value = state["count"]          # Raises KeyError if the column doesn't exist
value = state.get("count")      # Returns None if the column doesn't exist
exists = "count" in state       # Checks existence

The Payload class also provides:

• keys() — a list of column names with non-empty values.
• to_dict() — converts to a standard Python dict.

Main operations

Reading state

state = ctx.external_state("/join-state", message)

# Read a column value
hit_payload = state.get("hit_payload", str)
show_time = state.get("show_time")

# Dict-like access
try:
    value = state["hit_payload"]
except KeyError:
    value = None

Writing state with PayloadBuilder

To modify the state, use the to_builder() / set() / finish() pattern:

state = ctx.external_state("/join-state", message)

# Get a builder with current values
builder = state.to_builder()

# Update the required fields
builder.set("hit_payload", "some_value")
builder.set("show_time", 1234567890)

# Save the updated state
state.set(builder.finish())

The to_builder() method returns a PayloadBuilder pre-filled with the current state values. The builder.set(column, value) method returns the builder itself (it supports chaining calls). The builder.finish() method creates a new Payload and resets the builder.

Clearing state

state = ctx.external_state("/join-state", timer)

# Delete the row from the table
state.clear()

Configuration in the static spec

To use External State, you must declare an external state manager in the external_state_managers section of the computation in the static spec:

"computations" = {
    "reducer" = {
        "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
        "group_by_schema" = [
            {"name" = "hash"; "expression" = "farm_hash(key)"; "type" = "uint64"};
            {"name" = "key"; "type" = "string"};
        ];
        "input_stream_ids" = ["input"];
        "output_stream_ids" = ["output"];
        "external_state_managers" = {
            "/shuffle-state" = {
                "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
                "parameters" = {
                    "path" = "//path/to/state/table";
                };
            };
        };
        "parameters" = {};
    };
};

Key fields:

• external_state_managers — a top-level section inside Computation that describes external state managers (previously parameters/external_states).
• The key inside external_state_managers (for example, "/shuffle-state") — the state name you use in Python code when calling ctx.external_state("/shuffle-state", message). The name must start with /.
• external_state_manager_class_name — the name of the registered external state manager class. For a typical scenario, use "NYT::NFlow::TSimpleExternalStateManager"; For more details, see the C++ documentation.
• parameters.path — the path to the YTsaurus dynamic table where the state is stored.

Creating a table for the state

You must create the table for External State in advance. The schema of the table's key columns must match the computation's group_by_schema.

Complete example — EventReducer from Shuffle

Example from shuffle:

from yt.yt.flow.library.python.companion.computation import RowFunction


class EventReducer(RowFunction):
    """Counts the number of events for each key using external state."""

    def on_message(self, message, output, ctx):
        state = ctx.external_state("/shuffle-state", message)
        builder = state.to_builder()
        builder.set("count", (state.get("count") or 0) + 1)
        state.set(builder.finish())

Full source code

Work pattern:

1. Get the current state with ctx.external_state(...).
2. Create a builder with state.to_builder().
3. Update the required fields with builder.set(...).
4. Save with state.set(builder.finish()).

See also

• StateAccessor (Python)
• Internal State (Python)
• Working with states (Python) — a brief overview
• Stateful processing (concept)
• External State (Java)
• Shuffle example