Python Pipelines And Graphs

Python streams use the same split as Rust Datum:

• Source describes where elements come from.
• Flow describes reusable transformations.
• Sink describes the terminal.
• RunnableGraph is the materializable blueprint returned by to_mat().

All of these objects are immutable blueprints. Reusing the same graph starts a fresh materialization each time.

The shipped stubs are generic for the stream and port surface:

source: datum.Source[int] = datum.Source.range(0, 10)
flow: datum.Flow[int, int] = datum.Flow.map_add(1)
sink: datum.Sink[int, list[int]] = datum.Sink.collect()
graph: datum.RunnableGraph[datum.StreamCompletion[list[int]]] = source.via(flow).to_mat(sink)

Datum's Python CI runs both mypy and pyright against examples that must pass and examples that must fail, including mismatched via() chains, wrong-type sinks, and GraphBuilder.connect() calls with incompatible typed ports.

Linear DSL

import datum

pipeline = (
    datum.Source.from_iter([1, 2, 3, 4])
    .map_add(10)
    .drop(1)
    .take(2)
    .to_mat(datum.Sink.collect())
)

with datum.Runtime() as runtime:
    assert pipeline.run(runtime).wait() == [12, 13]
    assert pipeline.run(runtime).wait() == [12, 13]

Common source constructors:

Constructor	Use
Source.from_iter(values)	Finite integer stream from a Python iterable
Source.range(stop) / Source.range(start, stop, step)	Integer range source
Source.single(value)	One integer element
Source.from_arrow(data)	Arrow RecordBatch stream for batch UDFs

Common integer operators:

Operator	Use
map(function)	Callable convenience tier; runs as Arrow UDF batches over one integer column
filter(function)	Callable convenience tier; callable must return bool
flat_map(function)	Callable convenience tier; callable returns an iterable of int, flattened in order
map_add(n) / map_subtract(n) / map_multiply(n)	Named Rust arithmetic kernels
filter_equal(n) / filter_not_equal(n)	Named Rust equality kernels
filter_less_than(n) / filter_less_or_equal(n)	Named Rust ordered-comparison kernels
filter_greater_than(n) / filter_greater_or_equal(n)	Named Rust ordered-comparison kernels
take(n) / drop(n)	Bound or skip elements
via(flow)	Attach a reusable Flow

map(function), filter(function), and flat_map(function) are intentionally ergonomic, not the fastest tier. Datum groups integer elements into Arrow batches, crosses into Python once per batch, and runs the per-element loop inside the generated Python wrapper. The default batch size is chosen from the benchmark sweep; pass batch_size=... to tune it. Upstream completion flushes the final partial batch.

graph = (
    datum.Source.from_iter([1, 2, 3, 4])
    .map(lambda x: x + 10)
    .filter(lambda x: x > 12)
    .flat_map(lambda x: [x, x * 10])
    .to_mat(datum.Sink.collect())
)

For hot integer arithmetic and comparisons, use the named Rust kernels. For batch/vectorized work, use typed col() expressions when the operation is lowerable, or map_batches() with PyArrow kernels for arbitrary Python.

Named integer terminals avoid operation selectors:

Sink	Use
Sink.fold() / Sink.fold_sum(initial=0)	Sum elements from the initial value
Sink.fold_product(initial=1)	Multiply elements from the initial value
Sink.collect()	Collect elements into a list
Sink.for_each_count()	Count elements

Sink.fold() keeps the common sum spelling, but its former op= keyword has been removed.

Reusable Flows

flow = datum.Flow.map_add(1).via(datum.Flow.filter_greater_than(2))
graph = datum.Source.from_iter([1, 2, 3]).via(flow).to_mat(datum.Sink.for_each_count())

with datum.Runtime() as runtime:
    assert graph.run(runtime).wait() == 2

All named integer kernels are also reusable Flow factories. Flow.map(), Flow.filter(), and Flow.flat_map() build the same callable convenience tier as the corresponding Source methods. Flow.map_batches() is the separate vectorized Arrow-batch UDF factory.

Batch Expressions And Metrics

Arrow batch streams use datum.col() to build typed, schema-checked expressions:

import pyarrow as pa

batch = pa.record_batch([pa.array([1, 4, 8])], names=["value"])
batch_source = (
    datum.Source.from_arrow(batch)
    .with_column("scaled", datum.col("value") * 3)
    .filter(datum.col("scaled") >= 10)
)

report = batch_source.execution_report(metrics_level="timing")
snapshot = report.metrics_snapshot()

Expression schemas are validated when the blueprint is built. They lower to Arrow Rust kernels by default, and execution_report() exposes the requested and selected backend for every batch node. Metrics are always opt-in; enabled reports return immutable per-node snapshots. Use map_batches(udf, output_schema=...) instead when the batch operation needs arbitrary Python. See Arrow Expressions And UDFs for the full expression, backend, and metrics contracts.

GraphDSL Junctions

Use GraphDSL when a linear chain is not enough. The Python GraphDSL supports the acyclic junctions used by the first GraphDSL surface: Broadcast, Balance, Merge, Zip, Partition, Concat, and Interleave.

import datum

def balance_merge_graph():
    def build(builder):
        balance = builder.add(datum.Balance(2))
        merge = builder.add(datum.Merge(2))
        builder.connect(balance.outlet(0), merge.inlet(0))
        builder.connect(balance.outlet(1), merge.inlet(1))
        return datum.FlowShape(balance.inlet(), merge.outlet())

    return datum.GraphDsl.create(build)

graph = (
    datum.Source.range(0, 6)
    .via_graph(balance_merge_graph())
    .to_mat(datum.Sink.collect())
)

with datum.Runtime() as runtime:
    assert graph.run(runtime).wait() == [0, 1, 2, 3, 4, 5]

Partition currently exposes the modulo strategy. Omit the argument for the documented default, or select it explicitly through the typed surface:

partition = datum.Partition(4, strategy=datum.PartitionStrategy.MODULO)

Non-default raw strings fail at construction and name PartitionStrategy.MODULO as the replacement.

Shape accessors are explicit: fan-out shapes use .inlet() and .outlet(i), fan-in shapes use .inlet(i) and .outlet(), and Zip uses .in0(), .in1(), and .outlet(). Invalid wiring surfaces as a Datum exception at graph creation. GraphDsl.create() validates port ownership, duplicate wiring, element types, exposed shape ports, internal edges, and shape completeness before it returns the immutable graph blueprint.

Materialized Values And Keep

to_mat() chooses the sink and the materialized value. The default is datum.Keep.RIGHT, which returns the sink's StreamCompletion.

with datum.Runtime() as runtime:
    left, done = datum.Source.from_iter([1, 2]).to_mat(
        datum.Sink.collect(),
        keep=datum.Keep.BOTH,
    ).run(runtime)

    assert repr(left) == "NotUsed"
    assert done.wait() == [1, 2]

datum.Keep.LEFT, datum.Keep.RIGHT, datum.Keep.BOTH, and datum.Keep.NONE mirror Datum's materialized-value selection. In the current Python surface, the source side is usually NotUsed and the sink side is usually a StreamCompletion.

Completion Handles

run(runtime) returns immediately with the materialized value. For the usual Keep.RIGHT case that is a StreamCompletion; call wait() to block until the stream finishes and retrieve the terminal result.

StreamCompletion.wait() is one-shot. A second wait() on the same completion raises BuildError. If you need another result, materialize the blueprint again.

Runtime can be used as a context manager. Leaving the with block shuts it down.

Errors

Python exceptions follow a small hierarchy:

Exception	When it is raised
DatumError	Base class for Datum Python exceptions
BuildError	Blueprint construction errors, bad keep modes, removed compatibility calls, repeated wait()
StreamError	Runtime stream failures, arithmetic overflow, connect failures, UDF failures

Build-time validation is strict for graph wiring and declared schemas: connect(), via(), to_mat(), from_arrow(), map_batches(), and GraphDsl.create() fail at construction when the edge or blueprint is invalid. Runtime failures are reserved for execution behavior such as integer overflow, UDF exceptions, or a UDF returning a batch whose actual schema differs from the declared output_schema.