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()

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.

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().