--- metadata: - name: generator content: Diplodoc Platform v5.50.6 alternate: - https://ytsaurus.tech/docs/en/flow/concepts/stateful.md - https://ytsaurus.tech/docs/ru/flow/concepts/stateful.md --- > **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt # Stateful processing in YTsaurus Flow Use stateful processing to handle events with read-modify-write operations on states stored in YTsaurus. For example, you can count statistics for incoming events by key: you load the old value, update it, and write it back. ## State access model {#model} You access state inside a [computation](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation) through three components: - **State** — user data tied to a [key](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#key) and stored in a YTsaurus dynamic table. - **State client** — a type-specific object (`Client`) that the process function creates for each named state and initializes in `Init`. You access state by key through the client. It can be **read-write** (works with both [internal](#internal-state) and [external](#external-state) states) or **read-only** ([joiner](#external-state-joiner) for external state). - **State accessor** — what the client returns for a specific key (based on an input message, timer, or explicit key): a representation of the state of the same type `TState`. The accessor behaves like a smart pointer to the state: a read-write accessor lets you read, modify, and clear the state; a read-only accessor lets you only read it. {% note warning %} The accessor is valid only within the current [epoch](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#epoch). Don't store it in process-function fields or reuse it across epochs — get the state through the client again in each epoch. {% endnote %} ## State types {#state-types} ### Internal State {#internal-state} This is the simplest way to work with state: you don't need to create tables — Flow manages them automatically. Data loads at the start of the [epoch](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#epoch) and writes on commit. You get read-write access through the same client you use for external state. The state type can be arbitrary; the only requirement is that it's serializable to YSON (to persist between epochs). ### External State {#external-state} This is state in a user-managed dynamic table. You create and manage the tables. You get read-write access through the same client you use for internal state, but the backend is a **state manager** (`TSimpleExternalStateManager`). You declare it in the computation spec in the top-level `external_state_managers` section; the implementation resolves via `external_state_manager_class_name`. It supports caching. ### External State Joiner {#external-state-joiner} You get read-only access to external states via a key-based join — through a **joiner** (`TSimpleExternalStateJoiner`). You declare it in the computation spec in the top-level `external_state_joiners` section (at the same level as `external_state_managers`); the implementation resolves via `external_state_joiner_class_name`. It supports TTL-based caching: loaded states live in the shared [StateCache](#state-cache) and reload from YT only after the TTL expires or the state is evicted from the cache. {% note warning %} Tables accessed by a read-write state manager must be modified only through it (or when the [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) is [stopped](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#start-stop-pause-pipeline)), because it may use caches. {% endnote %} {% note warning "One table, one writer" %} Only one computation should write to an external state table: writes from different [partitions](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition) and transactions break state consistency. The state manager owns its table for writing: `TSimpleExternalStateManager` declares it as its own. To get read-only access to another computation's state, use an [external state joiner](#external-state-joiner) (`TSimpleExternalStateJoiner`) or send messages to the writer computation — joiners don't lock the table. The spec validation checks write ownership: each state table must have exactly one owner writer, and a pipeline where two managers lock the same table for writing is rejected with the error `State table is claimed for writing by both ...`. Read-only consumers (joiners) don't lock the table for writing, so they can share it with the owner writer. {% endnote %} {% note info %} For any state, an empty value corresponds to the absence of a row in the table. If the state is empty after modification, the corresponding row is deleted. By default, emptiness and state clearing are determined automatically (by comparing to the default value); the state type can override this behavior. {% endnote %} ## State storage {#storage} States are stored in YTsaurus dynamic tables. Here's a simple schema example: #| || **name** | **type** | **sort_order** | **expression** || || `hash` | `uint64` | `ascending` | `farm_hash(my_key)` || || `my_key` | `string` | `ascending` | || || `my_value_1` | `string` | | || || `my_value_2` | `string` | | || |# ### group_by_schema consistency {#group-by-schema} For correctness and performance, we strongly recommend that you use, as the [group_by_schema](https://ytsaurus.tech/docs/en/flow/concepts/spec.md#computation) for a [computation](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation), the schema of the first key columns of the dynamic table with states (strictly a prefix of the key columns). This ensures that: - Only one [partition](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition) handles events for a single key (correctness). - One partition handles a limited number of tablets (performance). Here's an example of a `group_by_schema` consistent with the state table schema from the example above: #| || **name** | **type** | **expression** || || `hash` | `uint64` | `farm_hash(my_key)` || || `my_key` | `string` | || |# ## StateCache {#state-cache} Flow provides a shared two-level (uncompressed + compressed) LRU cache for states. Configure it at `/dynamic_spec/job_tracker/state_cache`. Source: [yt/yt/flow/library/cpp/common/spec.h](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/library/cpp/common/spec.h) #| || **Parameter** | **Description** || || `compressed_cache_weight` | **Type**: [NYT::NYTree::TSize](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NYTree_TSize) **Default value**: `1Gi` Limit for `compressed`. || || `uncompressed_cache_weight` | **Type**: [NYT::NYTree::TSize](https://ytsaurus.tech/docs/en/flow/generated_docs/all_yson_structs.md#NYT_NYTree_TSize) **Default value**: `100Mi` Limit for `uncompressed`. || |# ## Implementation in different languages - **C++**: the client `TMutableStateKeyClient` (read-write) or `TJoinedStateKeyClient` (read-only) returns the accessor `TStateAccessor` / `TConstStateAccessor`; the same key client works with both internal and external states. [Learn more →](https://ytsaurus.tech/docs/en/flow/cpp/state.md) - **Java**: YsonStateAccessor, ProtoStateAccessor, ExternalStateAccessor. [Learn more →](https://ytsaurus.tech/docs/en/flow/java/state.md) - **Python**: ctx.state(), ctx.external_state(), ctx.proto_state(). [Learn more →](https://ytsaurus.tech/docs/en/flow/python/state.md) - **Go**: `flow.OpenYSONState`, `flow.OpenProtoState`, `flow.OpenExternalState`, and `flow.OpenJoinedExternalState` open a keyed state accessor. [Learn more →](https://ytsaurus.tech/docs/en/flow/go/state.md) ## See also - [Working with states (C++)](https://ytsaurus.tech/docs/en/flow/cpp/state.md) - [Working with states (Java)](https://ytsaurus.tech/docs/en/flow/java/state.md) - [Working with states (Python)](https://ytsaurus.tech/docs/en/flow/python/state.md) - [Working with states (Go)](https://ytsaurus.tech/docs/en/flow/go/state.md)