Working with states in YTsaurus Flow (Java)

The Java SDK Flow (Java and Kotlin) provides several types of state accessors for working with state. The most commonly used are:

• YsonStateAccessor — for YSON states stored in Flow’s internal tables.
• ExternalStateAccessor — for external states stored in separate dynamic tables in YTsaurus.

You also have access to ProtoStateAccessor, DefaultStateAccessor, RawStateAccessor, and NoOpStateAccessor. For more details on all types, see the Internal State section.

YsonStateAccessor

YsonStateAccessor<T> provides typed access to a YSON state tied to a message key. You get the accessor via RuntimeContext:

StateAccessor<T> stateAccessor = ctx.getYsonStateAccessor("state-name", message, StateClass.class);

val stateAccessor: StateAccessor<T> = ctx.getYsonStateAccessor("state-name", message, StateClass::class.java)

Parameters:

• "state-name" — the state name; it must match the name registered in the spec of the computation (parameters/internal_states).
• message — the current message from which you extract the grouping key.
• StateClass.class — the Java class for serializing or deserializing the state.

StateAccessor methods

Example: WordCountMapper

@Component
public class WordCountMapper implements RowFunction {

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        Word input = message.getPayload();

        StateAccessor<WordCountState> stateAccessor =
                ctx.getYsonStateAccessor("word-state", message, WordCountState.class);

        var state = stateAccessor.getOrDefault(new WordCountState(input.getWord(), 0));
        state.setCount(state.getCount() + 1);
        stateAccessor.set(state);
    }
}

@Component
class WordCountMapper : RowFunction {

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val input: Word = message.getPayload()

        val stateAccessor = ctx.getYsonStateAccessor("word-state", message, WordCountState::class.java)

        val state = stateAccessor.getOrDefault(WordCountState(input.word, 0))
        state.count = state.count + 1
        stateAccessor.set(state)
    }
}

In this example:

• You extract a Word object with the word field from the message.
• You get an accessor for the "word-state" state, which is tied to the current message’s key.
• If the state for this key doesn’t exist, you create a new WordCountState object with an initial counter value of 0.
• You increase the counter value and update the state.

Annotate the state class with @YTreeObject to serialize it to YSON:

@YTreeObject
public class WordCountState {
    private String word;
    private long count;

    public WordCountState() {}

    public WordCountState(String word, long count) {
        this.word = word;
        this.count = count;
    }

    // getters and setters
    public String getWord() { return word; }
    public void setWord(String word) { this.word = word; }
    public long getCount() { return count; }
    public void setCount(long count) { this.count = count; }
}

@YTreeObject
class WordCountState {
    var word: String = ""
    var count: Long = 0
    constructor()
    constructor(word: String, count: Long) {
        this.word = word
        this.count = count
    }
}

ExternalStateAccessor

ExternalStateAccessor provides access to an external state stored in a separate dynamic table in YTsaurus. You describe the external state with an ExternalStateDescriptor constant, which you create via StateDescriptors.external(...):

ExternalStateAccessor externalStateAccessor = ctx.getExternalStateAccessor("state-name", message);

val externalStateAccessor = ctx.getExternalStateAccessor("state-name", message)

ExternalStateAccessor methods

Payload is an untyped container with field access by name. Use PayloadBuilder to modify it.

Example: EventReducer

public class EventReducer implements RowFunction {

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        ExternalStateAccessor externalStateAccessor =
                ctx.getExternalStateAccessor("shuffle-state", message);

        Payload state = externalStateAccessor.getOrDefault();

        PayloadBuilder stateBuilder = state.toBuilder();
        if (state.get("count", Long.class) == null) {
            stateBuilder.set("count", 1L);
        } else {
            stateBuilder.set("count", state.get("count", Long.class) + 1);
        }

        externalStateAccessor.set(stateBuilder.finish());
    }
}

class EventReducer : RowFunction {

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val externalStateAccessor = ctx.getExternalStateAccessor("shuffle-state", message)

        val state = externalStateAccessor.getOrDefault()

        val stateBuilder = state.toBuilder()
        if (state.get("count", Long::class.java) == null) {
            stateBuilder.set("count", 1L)
        } else {
            stateBuilder.set("count", state.get("count", Long::class.java) + 1)
        }

        externalStateAccessor.set(stateBuilder.finish())
    }
}

In this example:

• You declare the SHUFFLE_STATE descriptor for the external state "/shuffle-state".
• You get the accessor for the message key via ctx.getState(SHUFFLE_STATE, message).
• You extract the current state value as Payload.
• You use PayloadBuilder to create an updated version of the state with an increased counter.
• You save the updated state back.

State in timers

When you process timers, the state is available via the timer object, which contains the grouping key:

@Override
public void onTimer(Timer timer, OutputCollector output, RuntimeContext ctx) {
    ExternalStateAccessor stateAccessor =
            ctx.getExternalStateAccessor("join-state", timer);

    Payload joinState = Objects.requireNonNull(stateAccessor.get());

    // process the state and generate output messages
    var messageBuilder = ctx.createMessageBuilder("output_stream");
    messageBuilder.set("hit_id", joinState.get("hit_id", String.class));
    // ... fill in the remaining fields ...
    output.addMessage(messageBuilder.finish());

    // clear the state after processing
    stateAccessor.clear();
}

override fun onTimer(timer: Timer, output: OutputCollector, ctx: RuntimeContext) {
    val stateAccessor = ctx.getExternalStateAccessor("join-state", timer)

    val joinState = stateAccessor.get()!!

    // process the state and generate output messages
    val messageBuilder = ctx.createMessageBuilder("output_stream")
    messageBuilder.set("hit_id", joinState.get("hit_id", String::class.java))
    // ... fill in the remaining fields ...
    output.addMessage(messageBuilder.finish())

    // clear the state after processing
    stateAccessor.clear()
}

The clear() method removes the state for the given key. Make sure to call it after closing the window or finalizing the processing to avoid accumulating outdated data.

Binding to group_by_schema

In TTransformCompanionComputation, the key used to access state is defined by group_by_schema in the computation spec. The state accessor extracts it from the message or timer.

TTransformOrderedSourceCompanionComputation does not support group_by_schema. Its internal state uses the source partition key, so messages from one partition share that state. See Computation (Java) for choosing a SourceComputation class.

For more details on group_by_schema and its impact on state handling, see the Stateful computations section.

State configuration in the spec

YSON state (internal_states)

You register YSON states in the computation spec via parameters/internal_states:

"computations" = {
    "mapper" = {
        "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
        "parameters" = {
            "internal_states" = ["word-state"];
        };
    };
};

External state

You register the external state in the external_state_managers section of Computation. The name must start with / and match the value passed to StateDescriptors.external(...):

"computations" = {
    "reducer" = {
        "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
        "external_state_managers" = {
            "/shuffle-state" = {
                "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
                "parameters" = {
                    "path" = "//path/to/state/table";
                };
            };
        };
        "parameters" = {};
    };
};

The external_state_manager_class_name field specifies the registered external state manager class. For a typical scenario, this is "NYT::NFlow::TSimpleExternalStateManager". For more details on available managers, see the C++ documentation.

You need to create the state table in advance.

See also

• Stateful computations
• Internal State (Java)
• External State (Java)
• Computation (Java)
• Getting started (Java)