Async Request in YTsaurus Flow (Python)
This example shows a two-component pipeline that implements asynchronous request processing. One computation routes events into requests and accumulates responses. The other performs computation without state. This is a Python implementation of a similar C++ example.
Structure
The pipeline includes two computations:
-
state(StateKeeperFunction) — a stateful computation that:- Accepts events from the
eventstream and generates requests to therequeststream with a uniquerequest_id. - Accepts responses from the
responsestream and accumulates the total length (total_length) in the external state.
- Accepts events from the
-
processor(RequestProcessorFunction) — a stateless computation that accepts requests from therequeststream and immediately returns a response (the length of the request string) to theresponsestream.
The event → request → response → state cycle closes between the two computations.
state_keeper_function.py
This file handles routing of incoming streams (event / response) and working with the external state.
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
This is a stateless request handler: it calculates the length of the request string and returns the response.
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
This is the entry point: it creates the pipeline and registers both computations.
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")
Key patterns
- Routing by
stream_id: theif stream_id == "event" / "response"branching lets a single computation handle multiple input streams with different logic. - Generating a unique
request_id:random.getrandbits(64)ensures correlation between the request and the response in the asynchronous cycle. - External state via
ctx.external_state("/state", message): theto_builder()/set()pattern supports cumulative updates to the external state. - Stateless computation:
RequestProcessorFunctiondoesn’t use state — it’s a pure transformation of a request into a response, which lets you scale it independently. - Two-component pipeline:
pipeline.add("state", ...)andpipeline.add("processor", ...)register the computations. The streams between them are described in the spec.