Async Request в YTsaurus Flow (Python)

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

Исходный код

Структура

Пайплайн состоит из двух компьютейшенов:

state (StateKeeperFunction) — stateful-компьютейшен, который:
- Принимает события из стрима event и порождает запросы в стрим request с уникальным request_id.
- Принимает ответы из стрима response и накапливает суммарную длину (total_length) во внешнем стейте.
processor (RequestProcessorFunction) — stateless-компьютейшен, который принимает запросы из стрима request и немедленно возвращает ответ (длину строки запроса) в стрим response.

Цикл event → request → response → state замыкается между двумя компьютейшенами.

`state_keeper_function.py`

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

class StateKeeperFunction(RowFunction):
    """Sends requests for events; accumulates total_length for responses."""

    def on_message(self, message, output, ctx):
        stream_id = message.stream_id

        if stream_id == "event":
            key = message.payload["key"]
            data = message.payload["data"]
            request_id = random.getrandbits(64)

            builder = ctx.message_builder("request")
            builder.set("request_id", request_id)
            builder.set("key", key)
            builder.set("request", data)
            output.add_message(builder.finish())
            log.debug("Send request (RequestId: %s)", request_id)

        elif stream_id == "response":
            length = message.payload["length"]
            request_id = message.payload["request_id"]
            log.debug("Received response (RequestId: %s)", request_id)

            state = ctx.external_state("/state", message)
            current = state.get("total_length")
            total = (current if current is not None else 0) + length
            builder = state.to_builder()
            builder.set("total_length", total)
            state.set(builder.finish())
        else:
            raise ValueError(f"Unknown stream_id: {stream_id}")

`request_processor_function.py`

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

class RequestProcessorFunction(RowFunction):
    """Stateless: returns length of request string as response."""

    def on_message(self, message, output, ctx):
        request_id = message.payload["request_id"]
        key = message.payload["key"]
        request = message.payload["request"]
        length = len(request)

        builder = ctx.message_builder("response")
        builder.set("request_id", request_id)
        builder.set("key", key)
        builder.set("length", length)
        output.add_message(builder.finish())
        log.debug("Processed request (RequestId: %s)", request_id)

`main.py`

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

def main():
    log.info("Starting companion execution")

    pipeline = Pipeline()
    pipeline.add("state", StateKeeperFunction())
    pipeline.add("processor", RequestProcessorFunction())

    log.info("Starting pipeline...")
    pipeline.run()
    log.info("Pipeline completed")

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

Маршрутизация по stream_id: ветвление if stream_id == "event" / "response" позволяет одному компьютейшену обрабатывать несколько входных стримов с разной логикой.
Генерация уникального request_id: random.getrandbits(64) обеспечивает корреляцию между запросом и ответом в асинхронном цикле.
External state через ctx.external_state("/state", message): паттерн to_builder() / set() для накопительного обновления внешнего стейта.
Stateless-компьютейшен: RequestProcessorFunction не использует стейт — чистое преобразование запроса в ответ, что позволяет масштабировать его независимо.
Двухкомпонентный пайплайн: pipeline.add("state", ...) и pipeline.add("processor", ...) регистрируют компьютейшены, стримы между которыми описываются в спеке.

См. также

Shuffle

Retryable Async Request