External State в YTsaurus Flow (Java)

External State — механизм работы с внешним состоянием (стейтом), хранящимся во внешней динамической таблице YTsaurus. Пользователь самостоятельно создаёт таблицу для хранения стейта на том же кластере, где развёрнут пайплайн.

Общие сведения о stateful-обработке описаны в разделе Stateful processing.

Обзор

External State в Java SDK Flow (Java и Kotlin) представлен классом ExternalStateAccessor, который предоставляет типизированный доступ к строкам внешней динамической таблицы. Стейт привязан к ключу сообщения (group_by_schema). Подробнее про StateAccessor и работу со стейтом: State Accessor.

Отличие от Internal State

Подробнее о Internal State см. Internal State.

В отличие от внутреннего стейта, внешний стейт не отслеживается: новое значение задаётся только через set(), а стейт из joiner-а доступен только на чтение.

Получение ExternalStateAccessor

External state описывается константой ExternalStateDescriptor, создаваемой через StateDescriptors.external(...). Дескриптор обычно объявляется один раз на класс RowFunction:

private static final ExternalStateDescriptor JOIN_STATE =
        StateDescriptors.external("/join-state");

ExternalStateAccessor для конкретного ключа получается через RuntimeContext (который наследует StatefulContext):

// Для сообщения
ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("state-name", message);

// Для таймера
ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("state-name", timer);

// Для сообщения
val stateAccessor = ctx.getExternalStateAccessor("state-name", message)

// Для таймера
val stateAccessor = ctx.getExternalStateAccessor("state-name", timer)

Параметры:

• дескриптор — ExternalStateDescriptor, имя которого совпадает с ключом в external_state_managers статической спеки и обязательно начинается с /;
• message / timer — сообщение или таймер, для ключа которого нужно получить стейт.

Объявление proto external state

Профильный стейт (NYT::NFlow::NProfileState::TProfileStateManager) хранит protobuf-сообщение. Опишите его через StateDescriptors.externalProto(name, TMessage.class) — или StateDescriptors.externalProtoReadOnly(...) для джойнера — и работайте с ним через ProtoStateAccessor. В отличие от обычного external state, раннеру нужно знать сообщение, чтобы описать стейт воркеру: он записывает в спеку descriptor_layout.descriptor_set и message_name из объявленного класса. Поэтому объявите стейт и пайплайну — через PipelineContext.registerState(...), если контекст собирается вручную, или в Spring через ComputationProvider.getStates() либо бин StateDescriptor<?>. Одной константы в функции раннеру недостаточно.

Основные операции

Чтение стейта

ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("join-state", message);

// Получение стейта: null, если значения нет
Payload maybeState = stateAccessor.get();
if (maybeState != null) {
    String value = maybeState.get("field_name", String.class);
}

// Получение стейта с дефолтным значением (пустой Payload со схемой стейта)
Payload state = stateAccessor.getOrDefault();

val stateAccessor = ctx.getExternalStateAccessor("join-state", message)

// Получение стейта: null, если значения нет
val maybeState = stateAccessor.get()
if (maybeState != null) {
    val value = maybeState.get("field_name", String::class.java)
}

// Получение стейта с дефолтным значением (пустой Payload со схемой стейта)
val state = stateAccessor.getOrDefault()

Запись стейта

ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("join-state", message);

// Получение текущего стейта и модификация через PayloadBuilder
PayloadBuilder builder = stateAccessor.getOrDefault().toBuilder();
builder.set("hit_payload", "some_value");
builder.set("show_time", 1234567890L);
stateAccessor.set(builder.finish());

val stateAccessor = ctx.getExternalStateAccessor("join-state", message)

// Получение текущего стейта и модификация через PayloadBuilder
val builder = stateAccessor.getOrDefault().toBuilder()
builder.set("hit_payload", "some_value")
builder.set("show_time", 1234567890L)
stateAccessor.set(builder.finish())

Очистка стейта

ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("join-state", timer);

// Очистка стейта (удаление строки из таблицы)
stateAccessor.clear();

val stateAccessor = ctx.getExternalStateAccessor("join-state", timer)

// Очистка стейта (удаление строки из таблицы)
stateAccessor.clear()

Конфигурация в статической спеке

Для использования External State необходимо объявить external state manager в секции external_state_managers Computation в статической спеке:

"computations" = {
    "join" = {
        "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
        "group_by_schema" = [
            {"name" = "hash"; "expression" = "farm_hash(hit_id)"; "type" = "uint64"};
            {"name" = "hit_id"; "type" = "string"};
            {"name" = "hit_time"; "type" = "uint64"};
        ];
        "input_stream_ids" = ["action"; "hit"];
        "output_stream_ids" = ["joined_action"];
        "external_state_managers" = {
            "/join-state" = {
                "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
                "parameters" = {
                    "path" = "//path/to/state/table";
                };
            };
        };
        "parameters" = {
            "wait_for_actions" = "10s";
        };
    };
};

Ключевые поля:

• external_state_managers — секция с описанием внешних состояний (top-level поле компьютейшена, на одном уровне с parameters, не вложенное в него).
• Ключ внутри external_state_managers (например, "/join-state") — имя стейта, обязательно начинающееся с /. То же имя передаётся в дескриптор StateDescriptors.external("/join-state"), через который Computation получает аксессор: ctx.getState(JOIN_STATE_DESCRIPTOR, message).
• external_state_manager_class_name — полное имя зарегистрированного класса state manager'а; по умолчанию "NYT::NFlow::TSimpleExternalStateManager".
• parameters/path — путь к динамической таблице YTsaurus, в которой хранится стейт.

Read-only joiner

Если компьютейшену нужно только читать external state, которым владеет (пишет) другой компьютейшен, используется read-only joiner. Joiner не является владельцем таблицы стейта: он лишь джойнит её строки по ключу сообщения. Это позволяет нескольким компьютейшенам читать одну таблицу, сохраняя единственного писателя.

Дескриптор создаётся через StateDescriptors.externalReadOnly(...), а аксессор — через тот же ctx.getState(...):

private static final JoinedExternalStateDescriptor RATING_SETTINGS =
        StateDescriptors.externalReadOnly("/rating-settings");

ReadOnlyExternalStateAccessor accessor = ctx.getState(RATING_SETTINGS, message);
Payload settings = accessor.getOrDefault();

private val ratingSettings = StateDescriptors.externalReadOnly("/rating-settings")

val accessor = ctx.getState(ratingSettings, message)
val settings = accessor.getOrDefault()

ReadOnlyExternalStateAccessor поддерживает get() и getOrDefault(); вызовы set(...) и clear() бросают UnsupportedOperationException, так как joiner не владеет таблицей.

В статической спеке joiner объявляется в top-level секции external_state_joiners (на одном уровне с external_state_managers); имя должно совпадать с аргументом externalReadOnly(...):

"external_state_joiners" = {
    "/rating-settings" = {
        "external_state_joiner_class_name" = "NYT::NFlow::TSimpleExternalStateJoiner";
        "parameters" = {
            "path" = "//path/to/state/table";
        };
    };
};

Создание таблицы для стейта

Таблица для External State должна быть создана заранее. Схема ключевых колонок таблицы должна совпадать с group_by_schema Computation. Пример схемы:

Полный пример

Пример из wait_click_join:

public class JoinProcessFunction implements RowFunction {

    @Override
    public void onMessage(ExtendedMessage message, OutputCollector output, RuntimeContext ctx) {
        var streamId = message.getStreamId();

        // Получение ExternalStateAccessor для текущего ключа
        ExternalStateAccessor stateAccessor = ctx.getExternalStateAccessor("join-state", message);

        // Чтение текущего стейта (или создание пустого)
        PayloadBuilder joinState = stateAccessor.getOrDefault().toBuilder();

        if ("hit".equals(streamId)) {
            Hit hit = message.getPayload();
            joinState.set("hit_payload", hit.getHitPayload());
        } else if ("action".equals(streamId)) {
            Action action = message.getPayload();
            if (Boolean.TRUE.equals(action.isClick())) {
                joinState.set("click_time", action.getActionTime());
            } else {
                joinState.set("show_time", action.getActionTime());
            }
        }

        // Сохранение обновлённого стейта
        stateAccessor.set(joinState.finish());

        // Установка таймера для закрытия ключа
        long maxTime = hitTime + waitTime;
        output.addTimer(maxTime, hitTime);
    }

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

        // Генерация выходного сообщения на основе стейта
        if (joinState.get("show_time", Long.class) != null
                && joinState.get("show_time", Long.class) != 0) {
            JoinedAction result = new JoinedAction();
            // ... заполнение полей из стейта ...
            output.addMessage(new Message("joined_action", result));
        }

        // Очистка стейта после обработки
        stateAccessor.clear();
    }
}

class JoinProcessFunction : RowFunction {

    override fun onMessage(message: ExtendedMessage, output: OutputCollector, ctx: RuntimeContext) {
        val streamId = message.getStreamId()

        // Получение ExternalStateAccessor для текущего ключа
        val stateAccessor = ctx.getExternalStateAccessor("join-state", message)

        // Чтение текущего стейта (или создание пустого)
        val joinState = stateAccessor.getOrDefault().toBuilder()

        if ("hit" == streamId) {
            val hit: Hit = message.getPayload()
            joinState.set("hit_payload", hit.getHitPayload())
        } else if ("action" == streamId) {
            val action: Action = message.getPayload()
            if (Boolean.TRUE == action.isClick()) {
                joinState.set("click_time", action.getActionTime())
            } else {
                joinState.set("show_time", action.getActionTime())
            }
        }

        // Сохранение обновлённого стейта
        stateAccessor.set(joinState.finish())

        // Установка таймера для закрытия ключа
        val maxTime = hitTime + waitTime
        output.addTimer(maxTime, hitTime)
    }

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

        // Генерация выходного сообщения на основе стейта
        if (joinState.get("show_time", Long::class.java) != null
                && joinState.get("show_time", Long::class.java) != 0L) {
            val result = JoinedAction()
            // ... заполнение полей из стейта ...
            output.addMessage(Message("joined_action", result))
        }

        // Очистка стейта после обработки
        stateAccessor.clear()
    }
}