Async Request в YTsaurus Flow (Go)

Пример пайплайна, реализующего асинхронный поход во внешний сервис: один компьютейшен превращает события в запросы и накапливает ответы во внешнем стейте, другой обслуживает запросы без стейта. Go-реализация того же сценария, что и C++ пример.

Исходный код

Структура

Компаньон обслуживает два компьютейшена, injector остаётся нативным сорсом из спеки:

state (stateKeeper) — stateful-компьютейшен, сгруппированный по key, который:
- принимает события из стрима event и порождает запрос в стрим request со случайным request_id;
- принимает ответы из стрима response и складывает суммарную длину (total_length) во внешний стейт /state.
processor (requestProcessor) — stateless-компьютейшен, сгруппированный по request_id: принимает запросы из стрима request и сразу отвечает длиной строки запроса в стрим response.

Цикл event → request → response → state замыкается между двумя компьютейшенами. Событие отвечается запросом, а не сразу результатом, поэтому обслуживающая сторона никогда не задерживает обработку: ответ приходит позже, отдельным сообщением, и только тогда стейт ключа сдвигается.

`main.go`

Точка входа: создание пайплайна и регистрация обоих компьютейшенов.

Исходный код: main.go


type eventMessage struct {
	flow.YSONMessage
	Key  uint64 `yson:"key"`
	Data string `yson:"data"`
}

type requestMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Request   string `yson:"request"`
}

type responseMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Length    int64  `yson:"length"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[eventMessage](eventStreamID),
		flow.NewYSONStream[requestMessage](requestStreamID),
		flow.NewYSONStream[responseMessage](responseStreamID),
	)
	pipeline.Add(
		flow.NewRowComputation("state", &stateKeeper{}),
		flow.NewRowComputation("processor", &requestProcessor{}),
	)

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "async_request: %v\n", err)
		os.Exit(1)
	}
}

`state_keeper.go`

Маршрутизация входных стримов (event / response) и работа с внешним стейтом.


type totalLengthState struct {
	TotalLength int64 `yson:"total_length"`
}

type stateKeeper struct{}

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

func (k *stateKeeper) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	switch msg.StreamID {
	case eventStreamID:
		return k.sendRequest(rt, msg, out)
	case responseStreamID:
		return k.accumulate(rt, msg)
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}
}

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

	request := flow.NewYSONMessage[requestMessage](requestStreamID)
	request.RequestID = rand.Uint64()
	request.Key = event.Key
	request.Request = event.Data

	encoded, err := flow.ConvertFrom(rt, request)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

func (*stateKeeper) accumulate(rt flow.Runtime, msg flow.ExtendedMessage) error {
	var response responseMessage
	if err := msg.ConvertTo(&response); err != nil {
		return err
	}

	state, err := flow.OpenExternalState(rt, totalLengthStateName, msg)
	if err != nil {
		return err
	}

	var total totalLengthState
	if _, err := state.ConvertTo(&total); err != nil {
		return err
	}
	total.TotalLength += response.Length
	return state.ConvertFrom(&total)
}

`request_processor.go`

Stateless-обработчик запросов: вычисляет длину строки запроса и возвращает ответ.


type requestProcessor struct{}

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

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

	response := flow.NewYSONMessage[responseMessage](responseStreamID)
	response.RequestID = request.RequestID
	response.Key = request.Key
	response.Length = int64(len(request.Request))

	encoded, err := flow.ConvertFrom(rt, response)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

Ключевые паттерны

Маршрутизация по msg.StreamID: switch по идентификатору входного стрима позволяет одному компьютейшену обрабатывать несколько входов с разной логикой. Неизвестный стрим — ошибка, а не молчаливое игнорирование.
Случайный request_id: rand.Uint64() связывает запрос с ответом. Запрос несёт ключ исходного события, поэтому ответ, партиционированный по request_id, возвращается к тому стейту, которому принадлежит поход.
Внешний стейт через flow.OpenExternalState(rt, "/state", msg): строка преобразуется в totalLengthState через ConvertTo, изменяется как структура и сохраняется через ConvertFrom.
Stateless-компьютейшен: requestProcessor не использует стейт и сгруппирован по request_id, а не по ключу события, поэтому запросы одного ключа расходятся по всем партициям и масштабируются независимо.
Зависимость стримов: streams_dependency в спеке объявляет, что request порождается из event — воркер учитывает это при продвижении вотермарков.

См. также

External State Join

Retryable Async Request