Python quickstart

import ubik, run streaming SQL in-process over libubik, and take results as pyarrow, pandas or polars.

The Python module runs the engine inside your process over libubik. There is nothing to deploy and nothing to connect to.

pip install ubik-sql

The distribution is ubik-sql; the module is ubik.

One query

ubik.stream() runs SQL over a source and hands back a result you read however you like.

import ubik

tbl = ubik.stream(
    "SELECT merchant, count(*) AS n "
    "FROM orders "
    "GROUP BY merchant, TUMBLE(event_time, INTERVAL '1 MINUTE')",
    from_="kafka://localhost:9092/orders",
).arrow()

Read the whole result with .arrow(), .df() or .pl(), iterate the object for record batches, or call .rows() for dicts.

A host DataFrame as a dimension

Arrow in, Arrow out. A pandas, polars or pyarrow table passed in tables enters through the Arrow C stream and becomes a lookup dimension the stream joins against.

import ubik
import pandas

merchants = pandas.DataFrame({"id": [10, 20], "region": ["EU", "US"]})

tbl = ubik.stream(
    "SELECT m.region, count(*) AS c "
    "FROM orders o JOIN merchants m ON o.merchant_id = m.id "
    "GROUP BY m.region, TUMBLE(o.event_time, INTERVAL '1 MINUTE')",
    from_="kafka://localhost:9092/orders",
    tables={"merchants": merchants},
).arrow()

A windowed dimension for a point-in-time join

A ubik windowed-aggregate changelog carries only window_start, and a window's value is complete only at its end. An as-of join on window_start would attach a value that folds in events after the probe row's own time. Wrap the frame with its window and the engine synthesizes the window_end / valid_from the join reads instead, and refuses a leaky as-of on window_start (PIT_LEAKAGE_RISK).

import ubik

tbl = ubik.stream(
    "SELECT o.merchant, o.amount, f.hourly_count "
    "FROM orders o ASOF JOIN feats f "
    "ON o.merchant = f.merchant AND o.event_time >= f.valid_from",
    from_="kafka://localhost:9092/orders",
    tables={"feats": ubik.windowed(hourly_counts, "1 hour")},
).arrow()

A second live stream (interval join)

Where tables binds a bounded dimension, streams binds a second live source for a stream-stream interval join, the in-process twin of the CLI's --stream and the sibling of tables. Pass streams={"b": "kafka://..."} (a file:// path also works) on stream() or pipeline(), and join it in the SQL on an equi key plus a time band. Its schema is inferred at start like from_; one stream in v1. The result is byte-identical to the CLI --stream path and to the DuckDB batch oracle.

import ubik

tbl = ubik.stream(
    "SELECT c.campaign, count(*) AS n "
    "FROM clicks c JOIN impressions i "
    "ON c.user_id = i.user_id "
    "AND i.ts BETWEEN c.ts - INTERVAL '5' SECOND AND c.ts + INTERVAL '5' SECOND "
    "GROUP BY c.campaign, TUMBLE(c.ts, INTERVAL '1 MINUTE')",
    from_="kafka://localhost:9092/clicks",
    streams={"impressions": "kafka://localhost:9092/impressions"},
).arrow()

Durable live pipelines

ubik.pipeline() tails a topic without end and survives a process restart from its checkpoint. A restart loses no event and double-counts none.

import ubik

Q = ("SELECT merchant, TUMBLE(ts, INTERVAL '1 MINUTE') AS w, count(*) AS c "
     "FROM orders GROUP BY merchant, TUMBLE(ts, INTERVAL '1 MINUTE')")

with ubik.pipeline(Q, from_="kafka://localhost:9092/orders",
                   checkpoint="~/.ubik/orders") as p:
    for window in p:          # yields each closed window as it emits
        react(window)         # a pyarrow RecordBatch

The checkpoint path is the identity

The first run creates it; a later run resumes from it automatically, after a clean exit or after a crash. Point two runs at the same path and the second continues the first rather than starting over.

A prepared point transform

ubik.transform() compiles a stateless SELECT once and returns a callable you apply to in-memory batches on demand. It runs on the caller's thread, so a request path pays the compile cost once and reuses it per batch.

import ubik
import pyarrow as pa

f = ubik.transform(
    "SELECT r.id, r.amount * fx.rate AS usd "
    "FROM req r JOIN fx ON r.ccy = fx.ccy",
    input={"req": pa.schema([("id", pa.int64()), ("amount", pa.float64()),
                             ("ccy", pa.string())])},
    tables={"fx": fx},
)

out = f(request_batch)   # pyarrow.Table, compiled once, reused per request

Validate before you run

ubik.validate() parses and binds a query against its source without running it, and returns the output schema. A mistake surfaces as the same UbikError a run would raise, but before you commit to a long pass. Handy while writing a query in a notebook.

import ubik

schema = ubik.validate(
    "SELECT country, TUMBLE(ts, INTERVAL '1 MINUTE') AS w, sum(amount) AS v "
    "FROM events GROUP BY country, TUMBLE(ts, INTERVAL '1 MINUTE')",
    from_="file:///data/events.ndjson",
)

Errors

Every failure raises ubik.UbikError, carrying .code, .message and .hint.

import ubik

try:
    ubik.stream(sql, from_="kafka://localhost:9092/orders").arrow()
except ubik.UbikError as e:
    print(e.code, e.hint)

The codes are the same ones the CLI prints, so an error found in a notebook reads identically in a pipeline log.

Requirements

pyarrow>=14 is the result surface and is installed with ubik-sql. pandas and polars are optional, imported lazily by .df() and .pl().

Python quickstart

On this page