Работа со стейтами в YTsaurus Flow (Go)

Аксессор стейта открывается свободной функцией flow.OpenXxxState(rt, name, input) с тремя аргументами:

• rt — flow.Runtime, второй аргумент любого обработчика (см. Process Function).
• name — имя стейта, объявленное в спеке компьютейшена.
• input — вход, к ключу которого привязывается аксессор: flow.ExtendedMessage, flow.Timer или flow.Visit. Все три реализуют интерфейс flow.Input, поэтому работа со стейтом в обработчике сообщения, таймера и визита выглядит одинаково.

Функции открытия — свободные, а не методы Runtime: YSON- и Proto-аксессоры параметризуются типом стейта, а методы в Go собственных типовых параметров не имеют.

Аксессор адресует стейт ровно одного ключа и живёт в пределах одного запроса. Воркеру уезжают только те записи, в которые компьютейшен писал: чтение стейта не отправляет его обратно.

YSON State

Самый простой способ работы со стейтом — YSON-формат. Стейт описывается обычной Go-структурой с тегами yson:

state, err := flow.OpenYSONState[wordCountState](rt, "word-state", msg)
if err != nil {
    return err
}

if state.Empty() {
    state.Value().Word = word
}
state.Value().Count++
return nil

flow.YSONState[T] предоставляет Empty(), Get() (*T, bool), Value() *T, Clear() и ReadOnly(). Value() возвращает изменяемое значение, а для отсутствующего стейта создаёт zero value. Отдельный Set не нужен: после успешной обработки батча SDK сериализует изменения автоматически; при ошибке обработчика они отбрасываются.

type wordCountState struct {
	Word  string `yson:"word"`
	Count int64  `yson:"count"`
}

type wordCountMapper struct{}

var _ flow.RowFunction = (*wordCountMapper)(nil)

func (*wordCountMapper) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var input wordMessage
	if err := msg.ConvertTo(&input); err != nil {
		return err
	}

	state, err := flow.OpenYSONState[wordCountState](rt, wordStateName, msg)
	if err != nil {
		return err
	}

	fresh := state.Empty()
	counter := state.Value()
	if fresh {
		counter.Word = input.Word
	}
	counter.Count++
	return nil
}

Стейт привязан к ключу сообщения, определяемому через group_by_schema в спеке компьютейшена. Для каждого уникального ключа хранится независимое значение.

Raw State

Для хранения стейта в виде сырых байтов:

state, err := flow.OpenRawState(rt, "raw-state", msg)

Возвращает flow.RawStateAccessor с методами:

• Get() ([]byte, bool) — получить значение и признак его наличия.
• Or(fallback []byte) []byte — получить значение или fallback.
• Set(data []byte) error — сохранить значение.
• Clear() error — удалить стейт.

YSON- и Proto-стейты — обёртки над сырым аксессором: сам RawStateAccessor нужен, когда сериализацию компьютейшен выполняет самостоятельно.

Proto State

Для хранения стейта в виде Protobuf-сообщения:

state, err := flow.OpenProtoState[TJoinState](rt, "join-state", msg)

Тип стейта называется по значению (TJoinState), а стейт работает с указателем на него (*TJoinState) — именно на указателе сгенерированный код реализует proto.Message.

Возвращает *flow.ProtoState[T, PT] с методами:

• Empty() bool — проверить, отсутствует ли значение.
• Get() (PT, bool) — вернуть изменяемое сообщение.
• Or(fallback PT) PT — вернуть сохранённое сообщение или записать и вернуть fallback.
• Set(value PT) error — заменить значение целиком.
• Clear() — удалить стейт.
• ReadOnly() flow.ReadOnlyProtoState[T, PT] — представление только на чтение.

Сообщение изменяется на месте: изменения записываются по окончании батча без Set, см. Изменение значения на месте.

state, err := flow.OpenProtoState[TJoinState](rt, "join-state", msg)
if err != nil {
    return err
}

window := state.Or(&TJoinState{})
window.HitPayload = payload

return nil

External State

Внешний стейт — строка пользовательской динамической таблицы. В пользовательском коде она представляется обычной Go-структурой с тегами yson:

state, err := flow.OpenExternalState(rt, "/shuffle-state", msg)

Имя стейта — абсолютный путь, совпадающий с ключом в секции external_state_managers статической спеки. Имя без ведущего / отвергается ошибкой flow.ErrInvalidStateName, необъявленное имя — ошибкой flow.ErrUnknownState.

Возвращает flow.ExternalStateAccessor. Основные операции:

• ConvertTo(&value) (bool, error) — заполнить структуру сохранённой строкой; bool отличает отсутствующую строку.
• ConvertFrom(&value) error — сохранить поля структуры в строку стейта.
• Clear() error — удалить строку.

Низкоуровневые Get, Or, Builder, Set и Schema нужны только для динамических схем и поколоночной обработки.

Пример из Shuffle:

type shuffleState struct {
	Count int64 `yson:"count"`
}

type eventReducer struct{}

var _ flow.RowFunction = (*eventReducer)(nil)

func (*eventReducer) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	state, err := flow.OpenExternalState(rt, shuffleStateName, msg)
	if err != nil {
		return err
	}

	var counter shuffleState
	if _, err := state.ConvertTo(&counter); err != nil {
		return err
	}
	counter.Count++
	return state.ConvertFrom(&counter)
}

Паттерн работы с внешним стейтом:

1. Открыть стейт через flow.OpenExternalState(...).
2. Преобразовать строку в структуру через state.ConvertTo(&value).
3. Изменить поля структуры и сохранить её через state.ConvertFrom(&value).

Подробнее — в разделе External State (Go).

Joined External State

Если внешний стейт нужен только на чтение — например, для обогащения сообщений справочником, — компьютейшен объявляет его в секции external_state_joiners и открывает отдельной функцией:

reference, err := flow.OpenJoinedExternalState(rt, "/reference", msg)

Возвращает flow.JoinedExternalStateAccessor. Строка читается в структуру через ConvertTo(&value); записи у аксессора нет по устройству, потому что заджойненный стейт заполняется из запроса и никогда не едет обратно. Низкоуровневые Get, Or и Schema доступны для динамических схем.

Пространства имён не пересекаются: стейт, которым компьютейшен владеет, недоступен через flow.OpenJoinedExternalState, и наоборот.

type referenceState struct {
	Name *string `yson:"name"`
}

type lookupJoin struct{}

var _ flow.RowFunction = (*lookupJoin)(nil)

func (*lookupJoin) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var event eventMessage
	if err := msg.ConvertTo(&event); err != nil {
		return err
	}

	name, ok, err := referenceName(rt, msg)
	if err != nil || !ok {
		return err
	}

	enriched := flow.NewYSONMessage[enrichedMessage](enrichedStreamID)
	enriched.Key = event.Key
	enriched.Name = name
	encoded, err := flow.ConvertFrom(rt, enriched)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

func referenceName(rt flow.Runtime, msg flow.ExtendedMessage) (string, bool, error) {
	reference, err := flow.OpenJoinedExternalState(rt, referenceStateName, msg)
	if errors.Is(err, flow.ErrStateNotRead) {
		return "", false, nil
	}
	if err != nil {
		return "", false, err
	}

	var state referenceState
	exists, err := reference.ConvertTo(&state)
	if err != nil || !exists || state.Name == nil {
		return "", false, err
	}
	return *state.Name, true, nil
}

Исходный код примера

Стейт в таймерах

API работы со стейтом в обработчике таймеров идентичен — вместо сообщения передаётся таймер:

state, err := flow.OpenExternalState(rt, "/join-state", timer)

Пример из WaitClickJoin — окно закрывается по таймеру, и стейт очищается сразу после публикации результата:

func (*joinFunction) OnTimer(
	ctx context.Context,
	rt flow.Runtime,
	timer flow.Timer,
	out flow.OutputCollector,
) error {
	if timer.StreamID != timerStream {
		return fmt.Errorf("unhandled timer stream %q", timer.StreamID)
	}

	state, err := flow.OpenExternalState(rt, joinStateName, timer)
	if err != nil {
		return err
	}
	var window joinState
	if _, err := state.ConvertTo(&window); err != nil {
		return err
	}

	if window.ShowTime != nil && *window.ShowTime != 0 && window.HitPayload != nil {
		var key joinKey
		if err := timer.Key.ConvertTo(&key); err != nil {
			return err
		}
		joined, err := joinedAction(rt, key, window)
		if err != nil {
			return err
		}
		out.AddMessage(joined)
	}

	return state.Clear()
}

Стейт, очищенный в этом запросе, дальше читается как отсутствующий: компьютейшен видит стейт таким, каким он станет после ответа воркеру.

Привязка стейта к ключу

В TTransformCompanionComputation стейт привязывается к ключу сообщения, определяемому через group_by_schema в спеке компьютейшена. Все сообщения с одинаковым ключом разделяют один стейт. Ключ, к которому привязан аксессор, берётся из входа — поэтому один обработчик не может случайно записать стейт чужого ключа.

В TTransformOrderedSourceCompanionComputation поле group_by_schema не поддерживается. Ключом внутреннего стейта служит ключ партиции источника, поэтому все сообщения одной партиции разделяют стейт. Подробнее о выборе класса для SourceComputation см. в разделе Computation (Go).

Подробнее о конфигурации ключей см. Stateful-вычисления.

Схема ключа доступна в обработчике через rt.KeySchema(), а сам ключ — через поле Key входа (msg.Key, timer.Key, visit.Key).

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

Стейт, не объявленный в спеке, открыть нельзя: функция открытия вернёт flow.ErrUnknownState со списком объявленных имён. Объявляются стейты в трёх разных местах описания компьютейшена:

• Внутренние стейты (YSON, Raw, Proto) — списком имён в parameters.internal_states. Имена произвольные, без ведущего /.
• Внешние стейты, которыми компьютейшен владеет, — в секции external_state_managers. Ключ секции задаёт имя стейта (абсолютный путь), а поле external_state_manager_class_name — зарегистрированный класс менеджера; для типового сценария это "NYT::NFlow::TSimpleExternalStateManager".
• Внешние стейты, которые компьютейшен только читает, — в секции external_state_joiners, рядом с external_state_managers.

"mapper" = {
    "computation_class_name" = "NYT::NFlow::NCompanion::TTransformCompanionComputation";
    "external_state_managers" = {
        "/shuffle-state" = {
            "external_state_manager_class_name" = "NYT::NFlow::TSimpleExternalStateManager";
            "parameters" = {
                "path" = "<cluster=cluster_name>//path/to/state";
            };
        };
    };
    "external_state_joiners" = {
        "/reference" = {
            "external_state_joiner_class_name" = "NYT::NFlow::TSimpleExternalStateJoiner";
            "parameters" = {
                "path" = "//path/to/current";
            };
        };
    };
    "parameters" = {
        "internal_states" = ["word-state"];
    };
};

Таблицы внутренних стейтов создаёт и обслуживает Flow, таблицу внешнего стейта пользователь создаёт сам. Подробнее — в разделах Internal State (Go) и External State (Go).

См. также

• State Accessor (Go)
• Internal State (Go)
• External State (Go)
• Computation (Go)
• Stateful-вычисления