Skip to content

Quick Start

This page walks through the basic shape of a Turbine application: subscribing to a topic, transforming batches, persisting state, and producing to an output topic. For installation and broker setup, see Installation.

A Minimal Application

from turbine import KafkaBroker, RecordBatch, Turbine

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka)

@app.subscribe(kafka.topic("input-topic"), output=kafka.topic("output-topic"))
def transform(batch: RecordBatch) -> RecordBatch:
    import pyarrow.compute as pc
    mask = pc.greater(batch.column("value"), 0)
    return batch.filter(mask)

app.run()

Handler Patterns

Blocking calls and external services

Handlers are synchronous. You may make blocking HTTP calls or use time.sleep() to pace an external service. Each partition has its own execution thread, separate from the API and cluster coordination: a waiting handler delays that partition's processing, output commits and timer callbacks. Other partitions keep running. Each partition adds an execution thread. With parallelism > 1, the partition still waits for all its shards before committing.

A checkpoint already scheduled from committed state can finish while a handler waits. The in-flight batch is saved only after it completes.

Set finite timeouts on external calls. Graceful shutdown and partition handoff wait for an in-flight callback to return; Turbine cannot cancel arbitrary Python code. A slow service can therefore increase Kafka lag and delay shutdown. Calls to external services are outside Turbine's exactly-once output guarantee, so make side effects idempotent if a replay could repeat them. CPU-heavy code still needs sufficient cores, and Python code on a GIL-enabled interpreter shares the GIL.

Stateless transform

Receives a batch, returns a batch. No state across batches.

@app.subscribe(kafka.topic("input"), output=kafka.topic("output"))
def transform(batch: RecordBatch) -> RecordBatch:
    return batch

Stateful transform

Add a second parameter to receive a per-partition key-value store (backed by RocksDB). State survives restarts via checkpoints.

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState

@app.subscribe(kafka.topic("events"), output=kafka.topic("counts"))
def aggregate(batch: RecordBatch, state: TurbineState) -> RecordBatch:
    total = state.get("total") or 0
    total += batch.num_rows
    state.put("total", total)
    return batch

TurbineState methods: get(key), put(key, value), delete(key), get_bytes(key), put_bytes(key, value).

For per-key state and timers on a subscription with partition_key=, a class handler can define process_key(key, rows, state) instead of process: the runtime groups the batch per key and state arrives already scoped on it. The lower-level state.scope(partition_key_value) returns the same typed TurbineStateView by hand. See State and timers for registration, cancellation and callback examples with multiple handlers per partition.

Sink (no output)

Omit output. The handler should return None.

@app.subscribe(kafka.topic("metrics"))
def monitor(batch: RecordBatch) -> None:
    print(f"Got {batch.num_rows} rows")

Class-based handler

A handler can also be declared as a class that exposes a process(batch[, state]) method. The class is instantiated once per shard before any batch arrives, and process is then called with the same arguments as the function form.

@app.subscribe(kafka.topic("events"), output=kafka.topic("counts"))
class Aggregate:
    def __init__(self) -> None:
        # Per-shard setup goes here: building expressions, opening windows,
        # precomputing lookup tables, etc.
        self._threshold = 100

    def process(self, batch: RecordBatch, state: TurbineState) -> RecordBatch | None:
        if batch.num_rows < self._threshold:
            return None
        return batch

The contract is the same as the function form: stateless if process takes one argument, stateful if it takes two; return a RecordBatch when output is set, return None for a sink. You'll need the class form when:

  • You want per-instance setup that runs before the first batch (precompiled expressions, dictionaries, classifier state).
  • You use windowing primitives (PersistentTumbling / PersistentSliding / PersistentSession) — they need to register themselves on self in __init__. See Windowing.
  • You use parallelism > 1 to scale CPU on a hot partition — the runtime needs one independent instance per shard for state isolation. See Configuration and Partitioning.
  • You want lifecycle hooks like on_recover_persistent_windows that ride on the class.

Class-based handlers can be instantiated with no constructor arguments. If you need parameters, set them on the class as defaults or read them from the environment inside __init__.

Explicit input schemas

Pass an Arrow schema to the input topic's schema= to keep columns and types stable across partitions and restarts. Declare only the fields you need to avoid decoding unused columns. Inferred schemas are cached too; an explicit declaration does not by itself promise faster JSON parsing.

import pyarrow as pa

Event = pa.schema([
    pa.field("id", pa.string(), nullable=False),
    pa.field("value", pa.int64(), nullable=False),
    pa.field("active", pa.bool_(), nullable=False),
])

@app.subscribe(kafka.topic("events", schema=Event), output=kafka.topic("out"))
def process(batch: RecordBatch) -> RecordBatch:
    return batch

Plain Pydantic models are also accepted with the optional turbine-fast[pydantic] dependency. Models declare types and nullability only; validators, constraints, aliases and non-null defaults are refused. See schema forms and limitations.

Step-by-Step Walkthrough

A complete stateful pipeline, end to end: a running total per Kafka partition, an output topic, and state that survives a restart. It assumes Turbine is installed and a broker is running — see Installation, whose compose stack the rpk commands below target. On a Kafka broker, kafka-topics.sh and kafka-console-producer.sh do the same jobs.

1. Create the topics

Turbine does not create your input and output topics — only the internal ones behind repartition_by= / combine_by=. Create both up front:

docker compose exec redpanda rpk topic create orders totals

2. Write the app

Save this as orders.py. It sums an amount column across every batch it sees, keeps the running total in RocksDB, and publishes one row per batch to totals:

import pyarrow as pa
import pyarrow.compute as pc

from turbine import KafkaBroker, RecordBatch, Turbine, TurbineState

Order = pa.schema([
    pa.field("id", pa.string(), nullable=False),
    pa.field("amount", pa.float64(), nullable=False),
])

kafka = KafkaBroker(bootstrap="localhost:9092")
app = Turbine(kafka, from_earliest=True)

@app.subscribe(kafka.topic("orders", schema=Order), output=kafka.topic("totals"))
def running_total(batch: RecordBatch, state: TurbineState) -> RecordBatch:
    total = state.get("total") or 0.0
    total += pc.sum(batch.column("amount")).as_py()
    state.put("total", total)
    print(f"{batch.num_rows} orders, running total {total}")
    return pa.RecordBatch.from_pydict({"total": [total], "orders": [batch.num_rows]})

app.run()

Two things worth noticing. The handler takes a second parameter, which is what makes the subscription stateful — Turbine opens a RocksDB store per partition and checkpoints it. And it is called once per batch, not per message: batch.num_rows is however many messages were ready, so the aggregate is computed by a columnar kernel rather than a Python loop.

State lives in a RocksDB under state_dir (/tmp/turbine by default) and is snapshotted to checkpoint_url in the background. Neither needs configuring to run locally; both do before you deploy — see Configuration.

Run it:

uv run python orders.py

The boot banner prints where state and checkpoints go, the topics it wired up, and this warning:

⚠ orders: subscribed without partition_key

That is expected here and worth understanding rather than silencing: this subscription keeps one value per Kafka partition, so nothing tells Turbine which column its state is keyed by, and it could not redistribute that state if you rescaled the topic. A real keyed aggregation passes partition_key="<column>"; see Partitioning. Pass partition_key=None explicitly when a per-partition value is genuinely what you want, and the warning goes away.

The app then sits there with no input yet. Leave it running.

3. Feed it

In another terminal:

docker compose exec redpanda bash -c 'printf "%s\n" \
  "{\"id\":\"a\",\"amount\":10.5}" \
  "{\"id\":\"b\",\"amount\":4.5}" \
  "{\"id\":\"c\",\"amount\":25}" | rpk topic produce orders'

The app prints a line per batch. Depending on timing you may see one batch of three or three batches of one — batching is opportunistic, and your handler must not assume a size.

4. Read the output

docker compose exec redpanda rpk topic consume totals -o start -n 1

The message body is the row the handler returned, serialized as JSON: {"total":40.0,"orders":3}.

5. Look at the state

The node serves its REST API on port 8400 while the app runs. Ask it what this app holds:

curl -sX POST http://localhost:8400/state \
  -H 'Content-Type: application/json' \
  -d '{"lane": "logical", "partition_key": ""}'

{"workers": [{"worker_id": "work-00-p00", "topic": "orders", "partition": 0,
              "truncated": false,
              "entries": [{"key": "total", "value": "40.0", "key_group": 0}]}]}

One worker per partition, each with its own store — the key you wrote, under the name you wrote it. lane: "logical" is what makes it readable: state is stored behind a binary keying header, so the default raw lane answers "\u0000\u0000\u0000total" for the same entry. The same view, plus topology, windows and the event feed, is in the web console at http://localhost:8400/console. See Console & API for the rest of the surface.

6. Restart it

Stop the app with Ctrl-C and start it again. Produce three more orders and watch the running total continue from 40.0 instead of restarting at zero. Two separate mechanisms are behind that. Restarting on the same machine reopens the same local RocksDB, which holds both the value and the offset it had consumed up to — so the app resumes rather than replaying the topic (which is also why from_earliest only applies when there is no recorded progress). Snapshots to checkpoint_url are what makes the same thing work on a different machine, or after that directory is gone: failover, rescaling, a fresh container.

See Crash recovery for what happens when the stop is not graceful, and Delivery guarantees for the duplicate-free variant.

Note what this walkthrough does not do: the state here is a single key per partition, not per entity. Keying state by a column — and the Kafka requirements that come with it — is Partitioning.

Where to go next

  • Windowing for time-bucketed aggregations.
  • Side outputs for sending rows somewhere other than output= — a declared split by rule, or extra destinations written from the handler.
  • Partitioning for the Kafka and partition_key requirements when using stateful subscriptions.
  • Inference to score every batch with an ONNX model, in the pipeline.
  • Deployment for production configuration, tuning, and cluster mode.