Async Request в YTsaurus Flow (C++)

Пайплайн демонстрирует паттерн асинхронных внешних запросов с использованием process functions. События поступают на вход, преобразуются в запросы, обрабатываются детерминированной функцией и результаты накапливаются в стейте.

Исходный код

Компоненты пайплайна

TStateKeeper

TStateKeeper реализует IProcessFunction и запускается адаптером TProcessFunctionComputation. Функция использует TSimpleExternalStateManager для работы с внешним стейтом и обрабатывает два вида входных сообщений:

Поток event: при получении события создает TRequestMessage с уникальным RequestId и отправляет его в поток request.
Поток response: при получении ответа обновляет стейт — суммирует total_length из всех полученных ответов.

Различение потоков происходит через message->StreamId.

TRequestProcessor

TRequestProcessor реализует IProcessFunction и запускается адаптером TProcessFunctionSwiftMapComputation, который не сохраняет входные и выходные сообщения в YTsaurus. Функция получает TRequestMessage, выполняет обработку (в данном примере — вычисляет длину запроса) и генерирует TResponseMessage.

Использование SwiftMap-адаптера обосновано тем, что обработка запроса является чистой функцией: при одинаковых входных данных всегда генерируется одинаковый результат.

Типы сообщений

TEventMessage — наследник TYsonMessage. Содержит поля Key и Data.
TRequestMessage — наследник TYsonMessage. Содержит поля RequestId, Key и Request.
TResponseMessage — наследник TYsonMessage. Содержит поля RequestId, Key и Length.

Все типы сообщений регистрируются через макрос YT_FLOW_DEFINE_YSON_MESSAGE.

Ключевой паттерн: цикл запрос-ответ

Основная идея данного примера — построение цикла запрос-ответ внутри пайплайна с использованием нескольких потоков:

events → TStateKeeper → request (генерация запроса)
request → TRequestProcessor → response (обработка запроса)
response → TStateKeeper → стейт (накопление результатов)

TStateKeeper одновременно является и потребителем событий, и потребителем ответов. Он использует input_stream_ids = ["event", "response"] и определяет тип входного сообщения по StreamId.

В спеке process functions связываются с адаптерами явно:

для state указаны computation_class_name = "NYT::NFlow::TProcessFunctionComputation" и processing_function = "NYT::NFlow::NExample::TStateKeeper";
для processor указаны computation_class_name = "NYT::NFlow::TProcessFunctionSwiftMapComputation" и processing_function = "NYT::NFlow::NExample::TRequestProcessor".

Управление стейтом

TStateKeeper использует TSimpleExternalStateManager для хранения суммы длин всех обработанных запросов. Клиент стейта (TMutableStateKeyClient<TSimpleExternalState>) привязывается в Init(const IRuntimeInitContextPtr&) через initContext->InitExternalStateClient(StateClient_, "/state"). Параметры стейта (path к таблице и т.п.) объявляются в секции external_state_managers спеки компьютейшена.

Функция main

В main регистрируются три потока, а process functions регистрируются через YT_FLOW_DEFINE_PROCESS_FUNCTION:

RegisterStream<TEventMessage>("event") — входные события
RegisterStream<TRequestMessage>("request") — запросы к процессору
RegisterStream<TResponseMessage>("response") — ответы от процессора

Исходный код

TRequestProcessor

void TRequestProcessor::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto request = context->ConvertToYsonMessage<TRequestMessage>(message);
    auto response = New<TResponseMessage>();
    response->RequestId = request->RequestId;
    response->Key = request->Key;
    response->Length = std::ssize(request->Request);
    output->AddMessage(context->ConvertToMessage(response));
    YT_TLOG_DEBUG("Processed request")
        .With("RequestId", request->RequestId)
        .With("FailedAttempts", 0);
}

TStateKeeper

void TStateKeeper::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitExternalStateClient(StateClient_, "/state");
}

void TStateKeeper::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    if (message->StreamId == "event") {
        auto event = context->ConvertToYsonMessage<TEventMessage>(message);
        auto request = New<TRequestMessage>();
        request->RequestId = RandomNumber<ui64>();
        request->Key = event->Key;
        request->Request = event->Data;
        output->AddMessage(context->ConvertToMessage(request));
        YT_TLOG_DEBUG("Send request")
            .With("RequestId", request->RequestId);
    } else if (message->StreamId == "response") {
        auto response = context->ConvertToYsonMessage<TResponseMessage>(message);
        YT_TLOG_DEBUG("Received response")
            .With("RequestId", response->RequestId);
        auto state = StateClient_.GetState(message->Key);
        i64 totalLength = state->GetColumnValue<std::optional<i64>>("total_length").value_or(0);
        totalLength += response->Length;
        TPayloadBuilder builder(state->Schema);
        builder.Set(totalLength, "total_length");
        state->Payload = builder.Finish();
    } else {
        THROW_ERROR_EXCEPTION("Unexpected stream_id")
            .With("stream_id", message->StreamId);
    }
}

См. также

Shuffle

Retryable Async Request