sql-flow

SQLFlow: DuckDB for Streaming Data.

Quickstart | Tutorials | | Documentation

SQLFlow is a stream processing engine that lets you define pipelines with just SQL. It consumes a stream, runs DuckDB SQL over each batch, and writes the result out. Think of it as a lightweight, single-binary Flink.

• Sources: Kafka, WebSockets, webhooks.
• Sinks: Kafka, ClickHouse, Iceberg, the console, or anything DuckDB can COPY to (PostgreSQL, S3, parquet, MotherDuck, DuckLake).
• Built on DuckDB and Apache Arrow: ~900k messages/sec on a laptop, in about a quarter GiB of memory.
• One Go binary, one Docker image, one YAML file per pipeline. No cluster.
• Coverage and invariant matrix: what is tested, and separately, what is proven — regenerated from the suites on every push.

Quick Start (Getting Started in 5 Minutes)

1. Build a binary (see Installation for prebuilt binaries and Docker):

make sqlflow

1. Validate a pipeline against test data, without a broker. dev invoke runs the config's handler over a JSONL fixture and prints the result:

./bin/sqlflow dev invoke dev/config/examples/basic.agg.mem.yml dev/fixtures/simple.json

{"city":"New York","city_count":28672}
{"city":"Baltimore","city_count":28672}

1. Start Kafka locally:

docker-compose -f dev/kafka-single.yml up -d

1. Publish test messages. The publisher is a Python script. uv reads uv.lock and builds the environment on first run:

uv run python cmd/publish-test-data.py --num-messages=10000 --topic="input-simple-agg-mem"

1. Start a Kafka consumer to watch the output:

docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-simple-agg-mem

1. Run the pipeline:

./bin/sqlflow run -c dev/config/examples/basic.agg.mem.yml --max-msgs=10000

The consumer prints one row per city:

{"city":"San Francisco","city_count":177}
{"city":"New York","city_count":236}
{"city":"Miami","city_count":203}
{"city":"Baltimore","city_count":180}

Installation

sqlflow reaches DuckDB through the Arrow ADBC driver manager, which dlopens libduckdb at runtime. The binary is not standalone: wherever you run it, that shared library has to be present. SQLFLOW_DUCKDB_LIB points at it; without that variable sqlflow looks in /opt/homebrew/lib/libduckdb.dylib on macOS and /usr/local/lib/libduckdb.so on Linux.

The pinned DuckDB version lives in one place, the DUCKDB_VERSION file.

Docker (no libduckdb setup)

The published image bakes in a matching libduckdb.so and sets SQLFLOW_DUCKDB_LIB, so nothing else is needed. Images are multi-arch (linux/amd64, linux/arm64):

docker run --rm \
  -v $(pwd)/dev:/tmp/conf \
  turbolytics/sql-flow:latest \
  dev invoke /tmp/conf/config/examples/basic.agg.mem.yml /tmp/conf/fixtures/simple.json

To build the image from the repo, make sqlflow-image — it prints the tag it built, derived from git describe.

Prebuilt binary

Release binaries are published for linux and macOS on amd64 and arm64. Download the one matching your platform, then install a matching libduckdb:

chmod +x sqlflow___
./scripts/install-libduckdb.sh /usr/local/lib
export SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so
./sqlflow___ version

scripts/install-libduckdb.sh always fetches the linux libduckdb.so, for the architecture it detects. It is for linux hosts and containers only. On macOS, brew install duckdb puts the library at the default path and no environment variable is needed.

From source

Requires Go 1.25+, a C toolchain (cgo is mandatory), and libduckdb.

# macOS
brew install duckdb
make sqlflow

# linux
./scripts/install-libduckdb.sh /usr/local/lib
export SQLFLOW_DUCKDB_LIB=/usr/local/lib/libduckdb.so
make sqlflow

The binary lands at bin/sqlflow.

How SQLFlow Works

A pipeline has three parts:

Source — Kafka, a WebSocket, or webhooks, modelled as a stream of messages.

Handler — DuckDB SQL executed over a batch of that stream: filter, aggregate, enrich, or drop data.

Sink — where the SQL result goes: Kafka, ClickHouse, Iceberg, the console, or — through the sqlcommand sink — anywhere DuckDB can write.

A config file names all three, plus optional commands run before the pipeline starts (attaching databases, creating tables) and optional tables the engine manages across the pipeline's lifetime:

SQLFlow Use-Cases

• Streaming Data Transformations: Clean data and types and publish the new data (example config).
• Stream Enrichment: Add data to an input stream and publish the new data (example config).
• Data aggregation: Aggregate input data batches to decrease data volume (example config).
• Tumbling Window Aggregation: Bucket data into arbitrary time windows (such as "hour" or "10 minutes") (example config).
• Run SQL against the Bluesky Firehose: Execute SQL against any websocket source, such as the Bluesky firehose (example config).
• Stream Data to Iceberg: Stream writes to an Iceberg catalog.
• Stream Data to ClickHouse: Insert stream processing outputs into ClickHouse (example config).
• Enrich Streams with Postgres Data: Query postgres during stream processing to enrich stream data.
• Sink Kafka to Postgres: Insert stream processing outputs into postgres.

CLI Reference

sqlflow [command]

sqlflow run

Runs the pipeline: consume, batch, execute SQL, sink, commit offsets.

sqlflow run  [flags]
sqlflow run -c  [flags]

--stats-json writes a small object, useful for CI assertions:

$ sqlflow run -c dev/config/examples/benchmark.structured.mem.yml \
    --max-msgs=2000 --stats-json=/tmp/stats.json
...
{"messages_consumed":2000,"num_errors":0}

sqlflow serve

Serves named SQL over HTTP. A serve config declares datasets, each a name and a fixed SQL statement with typed parameters. serve attaches the data through the config's commands, then answers each request by binding its parameters into that statement. Nothing in a request becomes SQL text.

sqlflow serve  [flags]
sqlflow serve -c  [flags]

A serve config is its own file, with commands and serve and no pipeline. sqlflow validate checks it against the serve schema, and sqlflow config example --serve prints every key. This one serves a Postgres view:

…

Parameters are string, integer, or timestamp. A timestamp is RFC 3339 with an offset; encode a + as %2B in a URL. An absent parameter binds NULL, so default it in SQL with coalesce. Every statement must use every declared parameter. A dataset has either sql or grains, and a request names a grain with ?grain=.

An integer param may declare min, max, or both. A request outside them is 400 invalid_param, and the message names the bounds. Nothing is clamped: a clamp would answer 200 with less than the caller asked for. /v1/datasets lists the bounds with the param.

      params:
        - {name: top, type: integer, min: 1, max: 20}

A dataset with grains can declare a time range instead, and let the server choose the grain:

…

A zoned timestamp is UTC. A decimal is a string of its exact digits. NaN and infinities are strings. truncated: true means max_rows cut the result.

elapsed_ms is the query. queued_ms is how long the request waited for a session, which is what rises when the pool is too small for the load.

Every route is GET, and /healthz also answers HEAD. A HEAD of a dataset would run its query, borrow a session and discard the rows, so it is refused with 405.

A token is an identifier, not a secret: a browser page ships it in plain sight. It names the caller in the request log, and deleting it revokes the caller.

Every refusal is {"error": {"code", "message"}}:

At start, serve prepares every statement, so a missing table or a syntax error stops the process with user.sql.invalid rather than failing the first request. rate_limit is reserved: the config accepts the key, and a non-zero value stops the process with user.config.serve_reserved.

What to know before you deploy it:

• Attach READ_ONLY. serve runs the dataset SQL as written. A write in it runs too, unless the attachment refuses.
• Aggregate in the backend. DuckDB pushes filters and projections into an attached Postgres and runs GROUP BY itself, so a rollup over raw rows copies every row to DuckDB first. sqlflow rollup keeps pre-aggregated tables in Postgres and generates the datasets that read them. A Postgres view with the GROUP BY also pushes the filter in, but re-aggregates the range on every request.
• Bound Postgres connections. An attachment opens up to pg_connection_limit connections, 64 by default. Set it low for a small database, as a command. The pool does not multiply it: with eight sessions scanning at once and a limit of four, the measured peak was four, so the connections are shared across sessions rather than opened per session.
• A timeout stops reading, not always the query. At the deadline the caller gets 504, and the reader stops at the next batch and is released, which is ADBC's equivalent of cancelling. An operator that runs long before yielding a batch still runs to the end, holding its session. Bound the part that is usually slow in the backend instead: a libpq connection string takes options='-c statement_timeout=30000', so an attached Postgres enforces its own ceiling.
• A pool serves requests. serve.pool.size sessions answer at once, four by default. A request waits for a free session, and that wait counts toward the dataset's timeout, so an exhausted pool answers 504 query_timeout. queued_ms in the response and sqlflow_serve_session_wait_seconds