Bytewax Runtime

loom.streaming.bytewax is the production adapter that turns a declarative StreamFlow into a real Bytewax dataflow.

For a runnable end-to-end reference implementation, see: dummy-loom-streaming.

What it does

  • compiles the flow

  • resolves config bindings

  • wires Kafka sources and sinks

  • applies With / WithAsync

  • runs observability hooks

Public entrypoint

from loom.streaming.bytewax import StreamingRunner

When to use it

Use the Bytewax adapter when you want:

  • topic-to-topic streaming

  • typed message envelopes

  • branch-aware fan-out

  • CollectBatch before WithAsync

  • runtime observability for batch and node lifecycle events

Flow contract

The clean authoring contract is:

from loom.streaming import CollectBatch, IntoTopic, Process, WithAsync

process = Process(
    CollectBatch(max_records=50, timeout_ms=2000),
    WithAsync(
        process=Process(
            # per-message async step
            ...
            IntoTopic(...),
        ),
        max_concurrency=50,
    ),
)

CollectBatch belongs before WithAsync when you want batch aggregation as an input shape. If you want to wait for a batch outside the async step, keep that logic in the parent flow, not inside the task itself.

Writing your own destination

IntoTopic and IntoTable are the destinations loom ships. A destination it does not ship is a frozen dataclass satisfying IntoSink: no base class, no registration, no loom import beyond the payload type. The compiler recognises it by structure and resolves its configuration from streaming.sinks.<name>.

from collections.abc import Sequence
from dataclasses import dataclass
from typing import Any, ClassVar

@dataclass(frozen=True)
class IntoJsonl:
    payload: type[IncidentEvent]
    name: str = "incidents"          # the key under streaming.sinks
    router_branch_safe: ClassVar[bool] = True

    def build_partition(
        self,
        config: Any,                 # your own resolved section
        worker_index: int,
        worker_count: int,
        bridge: Any = None,
        session_manager: Any = None,
    ) -> "JsonlPartition":
        return JsonlPartition(config["path"])


class JsonlPartition:
    def __init__(self, path: str) -> None:
        self._path, self._buffer = path, []

    def write_batch(self, items: Sequence[IncidentEvent]) -> None:
        self._buffer.extend(items)   # once per Bytewax epoch

    def close(self) -> None:
        flush(self._path, self._buffer)   # must be idempotent

Process(IntoJsonl(payload=IncidentEvent)) type-checks under mypy --strict: the three attributes are read-only in the protocol, so a frozen dataclass, a msgspec struct or a class variable all satisfy them, and config and session_manager are opaque, so your sink narrows them to whatever its backend resolves. write_batch runs per epoch, per worker; close may be called even when write_batch never was.

Booting from YAML (resolvers=)

StreamingRunner.from_yaml(flow, path) loads the config and builds the runner; run(config_path=...) does the same load on an existing runner. Both register loom’s built-in AWS resolvers, secrets (Secrets Manager) and ssm (SSM Parameter Store), by default, so a flow booted through the factory reads a secret with no code beyond the factory call:

# config/streaming.yaml
kafka:
  consumer:
    brokers:
      - ${ssm:/prod/incidents/kafka-broker}
    group_id: incidents-service
    topics: ["incidents.in"]
    security:
      protocol: SASL_SSL
      sasl_mechanism: SCRAM-SHA-512
      sasl_username: incidents
      sasl_password: "${secrets:/prod/incidents/kafka-password}"

from loom.streaming.bytewax import StreamingRunner

runner = StreamingRunner.from_yaml(incident_flow, "config/streaming.yaml")
runner.run()

The built-in resolvers use boto3’s default region and credential chain, and create their client only when a placeholder resolves: a YAML with no ${secrets:...} or ${ssm:...} never touches AWS and boots without boto3. When a placeholder does resolve and boto3 is missing, the error names the extra to install, loom-kernel[config-ssm].

resolvers= adds your own prefixes or overrides a built-in by name, on from_yaml and on run(config_path=...). Any object with a name and a resolve(key) -> object works:

class VaultResolver:
    name = "vault"

    def resolve(self, key: str) -> str:
        return read_vault_secret(key)

runner = StreamingRunner.from_yaml(
    incident_flow, "config/streaming.yaml", resolvers=[VaultResolver()]
)

A resolver you pass with the same name as a built-in wins; a built-in default never replaces a resolver already registered earlier in the process, so calling the factory more than once is safe. load_config registers no defaults, and resolvers passed to it explicitly replace an earlier registration of the same name.

Runtime notes

  • WithAsync executes one message per task, concurrently.

  • CollectBatch emits observable batch events.

  • Errors are routed through the explicit error wiring.

  • msg and payload are public expression roots for routing predicates.