Pipelines
ubik pipeline create, start, list, logs, stop and delete: named durable jobs, one process each, resumed from checkpoint across a restart.
A pipeline is a named, durable streaming job. It tails a Kafka topic live, checkpoints for exactly-once, survives restarts, and is stopped by name.
There is no daemon. Each pipeline is one engine process, detached into its own session so it outlives the CLI invocation, and the OS is the supervisor.
The six commands
ubik pipeline create orders --from kafka://broker/orders --to kafka://broker/agg \
"SELECT merchant, count(*) AS n FROM orders GROUP BY merchant, TUMBLE(ts, INTERVAL 1 MINUTE)"
ubik pipeline start orders
ubik pipeline list
ubik pipeline logs orders -f
ubik pipeline stop orders
ubik pipeline delete orders| Command | What it does |
|---|---|
create | Define a pipeline. Does not start it. |
start | Start or resume as a supervised background process. |
list | Status, Kafka lag, and each engine's last progress report. |
logs | The engine log: rows, progress, status, errors. |
stop | Stop a running pipeline. A clean checkpoint is taken first. |
delete | Delete a stopped pipeline: definition, checkpoint and log. |
A pipeline name must be alphanumeric, may contain - and _, and cannot start
with either.
Follow mode
A pipeline runs in follow mode. Windows close only as the event-time watermark advances, which is the minimum across the topic's partitions, never because the data so far ran out.
A one-shot ubik --from ... replay of the same topic closes every remaining
window at the end instead, so it can emit rows a pipeline holding for a quiet
partition never will. When that happens, list and logs show wm held by
with the partition to look at.
create
ubik pipeline create <name> <sql> [flags]The SQL's FROM names the source topic. Project the window function to emit
each window's start; see windows for the output column.
| Flag | Default | Meaning |
|---|---|---|
--from | required | Source Kafka topic, kafka://broker/topic. |
--to | the log | Sink for the aggregation changelog, kafka://broker/topic. |
--checkpoint-every | 100 | Checkpoint every N chunks, roughly 2048 rows each. |
--watermark-delay | none | Bounded out-of-orderness: close a window at end + interval, so a late row within the grace still counts. |
--dimension | none | Lookup-join build side, name=file://path.ndjson or name=kafka://broker/topic. Repeatable. |
--schema-registry | none | Confluent Schema Registry URL for an Avro or JSON topic. |
--schema | none | Local Avro schema (.avsc) for a raw-Avro topic with no registry. |
--progress-interval | 10 seconds | Emit a JSON progress line to the log every interval. 0 seconds disables. |
A --dimension file path is stored absolute, so a restart re-drains it. A Kafka
changelog dimension is followed live.
State on disk
Everything a pipeline owns lives under one directory, $UBIK_HOME/pipelines/<name>/,
where UBIK_HOME defaults to $HOME/.ubik.
| File | Contents |
|---|---|
pipeline.json | The definition: from, to, sql, checkpoint-every. |
checkpoint | The exactly-once checkpoint: aggregate state plus Kafka offsets. |
checkpoint.lock | The engine's flock. Two engines can never share the directory. |
engine.log | Captured stdout and stderr, as JSON lines. |
engine.pid | The running engine's pid. Advisory; the lock is the truth. |
supervisor.lock | Serializes start/stop/delete for this pipeline. |
The lock is the truth, the pid file is a convenience
The flock is held for the process lifetime and released by the kernel on any
exit, including kill -9. Every liveness answer double-checks it, so an engine
whose registration was lost is still visible, stoppable, and protected from
deletion.
The pid file is there for readable output. A stale one left behind by a hard
kill does not block start, because the lock is free; a genuine duplicate is
refused, because it is held.