SQLFlow

https://turbolytics.io/docs/sqlflow/
Admin
SQLFlow is a stream processing engine that embeds DuckDB and Apache Arrow to build and run...
Tokens:52,394
Snippets:403
Trust Score:5.2
Update:3 weeks ago
Show doc for...
Tokens:
Raw
### Windowed Pipeline Output Example

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example output from the windowed pipeline after the first full minute. Shows counts per kind for the bucket start time.

```json
{"count":2434,"iso_string":"2026-09-15T00:53:00","kind":"commit"}
{"count":18,"iso_string":"2026-09-15T00:53:00","kind":"account"}
{"count":15,"iso_string":"2026-09-15T00:53:00","kind":"identity"}
```

--------------------------------

### Start SQLFlow with Kafka and Postgres Join

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

Starts SQLFlow with the Kafka-to-Postgres join configuration. The command attaches Postgres and initializes the Kafka source; the log output shows the consumer loop starting and SQLFlow idling for messages.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v1.0.6 \
  run /tmp/conf/config/examples/kafka.postgres.join.yml
```

```log
INFO  Executing command step  {"name": "load extensions"}
INFO  Executing command step  {"name": "attach usersdb"}
INFO  initializing kafka source  {"topics": "[input-user-clicks-demo]", "group.id": "test", "auto.offset.reset": "earliest"}
INFO  consumer loop starting
```

--------------------------------

### Start Kafka with make start-backing-services

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Starts the local Kafka backing services required for the tutorial. Run this before publishing events.

```bash
make start-backing-services
```

--------------------------------

### healthz response for starting status

Source: https://turbolytics.io/docs/sqlflow/operations/running-in-production

Example response from GET /healthz showing a starting status with no commit yet.

```json
{"commit_age_seconds":7.94,"interval_seconds":3600,"reason":"no commit yet","status":"starting"}
```

--------------------------------

### Start Kafka with Docker Compose

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Starts a single-node Kafka instance using the provided docker-compose file.

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

--------------------------------

### Generate pipeline.yml from config example

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Pipes the config reference output to a file to use as a starting point for editing.

```bash
./bin/sqlflow config example > pipeline.yml
```

--------------------------------

### Run SQLFlow Pipeline with Configuration File

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start the SQLFlow pipeline using the configuration file for the stdout example. Mounts the config directory and runs with a maximum of 200 messages.

```bash
docker run -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /examples/bluesky/bluesky.kafka.transformed.yml --max-msgs=200
```

--------------------------------

### Start SQLFlow Pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Starts the SQLFlow container with environment variables for Postgres and Kafka, followed by the expected initialization logs.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/kafka.postgres.sink.yml
```

```text
INFO  serving http  {"addr": ":8000", "routes": ["/healthz"]}
INFO  initializing kafka source  {"topics": "[input-kafka-postgres-sink]", "group.id": "test", "auto.offset.reset": "earliest", ...}
INFO  consumer loop starting
```

--------------------------------

### Start Kafka consumer for output-bluesky-firehose topic

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Start a Kafka console consumer on the output topic to see the stream of Bluesky messages. Run this before starting the SQLFlow pipeline.

```bash
docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-bluesky-firehose
```

--------------------------------

### Start SQLFlow in Docker

Source: https://turbolytics.io/docs/sqlflow/tutorials/s3-parquet-sink

Starts SQLFlow in Docker using the given configuration file and environment variables. The documented run measured 194,590 messages per second on an Apple M1 Pro with SQLFlow, Kafka, and MinIO on one Docker network.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_S3_ENDPOINT=host.docker.internal:9000 \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_BATCH_SIZE=5000 \
    -e SQLFLOW_SINK_BASEPATH=s3://sqlflow/test-sink-1 \
    turbolytics/sql-flow:v1.0.6 run /examples/s3.parquet.sink.yml --max-msgs=100000
```

--------------------------------

### Start Postgres container on Docker network

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Creates a Docker network named 'sqlflow' and starts a Postgres 16 container named 'sqlflow-postgres' on that network with password 'postgres'. This is the first step to set up the database for the tutorial.

```bash
docker network create sqlflow
docker run -d --name sqlflow-postgres --network sqlflow \
  -e POSTGRES_PASSWORD=postgres postgres:16
```

--------------------------------

### Pipeline startup log output

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Example log output from running the pipeline. The warning about not deriving reference tables is harmless because the handler is an INSERT, not a SELECT.

```log
INFO  Executing command step  {"name": "declare the post schema"}
INFO  Creating managed table  {"name": "posts_per_minute_by_lang"}
WARN  could not derive reference tables from handler sql  {"error": "handler sql did not parse: Only SELECT statements can be serialized to json!"}
INFO  consumer loop starting
INFO  starting watermark manager  {"table": "posts_per_minute_by_lang", "poll_interval": "10s"}
INFO  throughput  {"messages_consumed": 142, "total_throughput_per_second": 30.8}
```

--------------------------------

### Start Kafka console consumer

Source: https://turbolytics.io/docs/sqlflow/tutorials/intro

Starts a Kafka console consumer to view the SQLFlow output on the topic 'output-simple-agg-mem'.

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

--------------------------------

### Install SQLFlow and clone repository

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Commands to clone the SQLFlow repository, pull the required Docker image (v2026.09.14.1 or later), and install dependencies with `uv sync`. SQLFlow itself needs no Python.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker pull turbolytics/sql-flow:v2026.09.14.1
uv sync
```

--------------------------------

### Start Kafka Consumer for Windowed Output

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start a Kafka console consumer for the output topic of the windowed pipeline.

```bash
docker exec -it kafka1 kafka-console-consumer --bootstrap-server=kafka1:9092 --topic=output-bluesky-post-kind-count-1-minute
```

--------------------------------

### Start SQLFlow Tumbling Window Pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Docker command to start the SQLFlow tumbling window pipeline with the windowed configuration file. Sets the Kafka brokers environment variable.

```bash
docker run -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  -v $(pwd)/dev/config/examples:/examples \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /examples/bluesky/bluesky.kafka.windowed.yml
```

--------------------------------

### Run SQLFlow pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Starts the SQLFlow pipeline with the example config. --max-msgs stops the run after consuming that many messages; leaving it off runs as a daemon tailing the topic until stopped.

```bash
./bin/sqlflow run --metrics=prometheus dev/config/examples/kafka.motherduck.yml --max-msgs=1000000
```

--------------------------------

### Initialize SQLite Iceberg catalog and test table

Source: https://turbolytics.io/docs/sqlflow/tutorials/iceberg-sink

Runs the setup script to create the Iceberg catalog and test tables. Output shows the created tables and completion message. The catalog and warehouse land in /tmp/sqlflow/warehouse.

```bash
uv run python3 cmd/setup-iceberg-local.py setup
```

```text
created default.city_events
created default.bluesky_post_events
Catalog setup complete.
```

--------------------------------

### Install SQLFlow

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Clones the SQLFlow repository and builds the binary, which lands at bin/sqlflow. Requires the local development toolchain. To run the container instead, substitute docker run turbolytics/sql-flow:v1.0.6 for ./bin/sqlflow in the run step.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
make sqlflow
```

--------------------------------

### Common Log Format example line

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Example of a single Common Log Format line from the NASA trace files, showing the format of each request.

```text
199.72.81.55 - - [01/Jul/1995:00:00:01 -0400] "GET /history/apollo/ HTTP/1.0" 200 6245
```

--------------------------------

### SQLFlow config reference output

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Example output of the config reference command, showing the structure and comments for each option.

```yaml
# Main pipeline configuration.
pipeline:
  # Name of the pipeline.
  name: <string>
  # Description of the pipeline.
  description: <string>
  # Number of messages processed in a batch.
  batch_size: <integer>
  # Time interval to flush batches in seconds.
  flush_interval_seconds: <integer>
  # Longest a shutdown may take after SIGTERM. The final batch, the
  # managers' final poll and the state syncs share it. Absent means 30.
  # When it passes the process exits 15, and the next start replays what
  # was not written.
  drain_deadline_seconds: <integer>
  # Error handling strategy.
  on_error:
    # Defines how errors should be handled.
    policy: RAISE | IGNORE | DLQ
  ...
```

--------------------------------

### SQLFlow Stdout Output Example

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example output from the SQLFlow pipeline when writing to stdout. The console sink sorts keys, so 'kind' precedes 'time'.

```json
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
{"kind":"commit","time":"2026-09-15T01:04:48"}
```

--------------------------------

### View SQLFlow logs

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Check the container logs to see the pipeline startup and idle state. The log shows the service attached and the consumer loop starting.

```bash
docker logs sqlflow-tiger
```

--------------------------------

### Install libduckdb on macOS and Linux

Source: https://turbolytics.io/docs/sqlflow/introduction/local-development

Install libduckdb via Homebrew on macOS or via the provided script on Linux. On Linux, set SQLFLOW_DUCKDB_LIB to the library path. Without it, SQLFlow uses default paths. The script always fetches the Linux library, so it won't work on macOS.

```bash
# macOS: installs to the default path, no environment variable needed
brew install duckdb

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

--------------------------------

### Start SQLFlow container

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Run SQLFlow in the background with the name sqlflow-tiger. The -d flag detaches, and SQLFLOW_KAFKA_BROKERS is set to host.docker.internal:29092 because the compose file advertises that listener for container access.

```bash
docker run -d --name sqlflow-tiger -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_TIGERDATA_DSN \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v1.2.0 \
  run /tmp/conf/config/examples/published/tigerdata.enrich.yml
```

--------------------------------

### Publish 200k test messages to Kafka

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Run the publish-test-data.py script to send 200,000 user_action messages to the input-kafka-postgres-sink topic with the user_action fixture. This is the first step before starting SQLFlow.

```bash
uv run python3 cmd/publish-test-data.py --num-messages=200000 --topic="input-kafka-postgres-sink" --fixture=user_action
```

--------------------------------

### Run the SQL Flow pipeline with Docker

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Clones the sql-flow repository and runs the Docker container to execute the Bluesky Postgres windowed example configuration. The command mounts the local dev directory to /tmp/conf and uses the turbolytics/sql-flow:v2026.09.14.1 image.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker run --rm --network sqlflow \
  -v $(pwd)/dev:/tmp/conf \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/bluesky/bluesky.postgres.windowed.yml
```

--------------------------------

### Install SQLFlow: clone repo and pull Docker image

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-join

Clones the SQLFlow repository, changes into the directory, and pulls the v1.0.6 Docker image. Requires v1.0.5 or later; the command pulls v1.0.6.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow
docker pull turbolytics/sql-flow:v1.0.6
```

--------------------------------

### Run archive pipeline with sql-flow

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Runs the archive pipeline from the sql-flow repository. It reads the Kafka topic and writes large batches for Parquet object size. Requires the published image and the example config file.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_S3_ENDPOINT=host.docker.internal:9000 \
    turbolytics/sql-flow:v2026.09.19.1 run /examples/logs.archive.parquet.yml
```

--------------------------------

### Run rollup pipeline with sql-flow

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Runs the rollup pipeline from the sql-flow repository. It reads the same Kafka topic and writes to ClickHouse, shaped by its window. Requires the published image and the example config file.

```bash
docker run \
    -v $(pwd)/dev/config/examples:/examples \
    -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
    -e SQLFLOW_CLICKHOUSE_DSN=clickhouse://host.docker.internal:8123/test \
    turbolytics/sql-flow:v2026.09.19.1 run /examples/logs.rollup.clickhouse.yml
```

--------------------------------

### Query newest bucket from Postgres

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Queries the newest bucket from the posts_per_minute_by_lang table and shows the resulting output. The bucket closes 60 seconds after its end, so the first full minute appears about three minutes after the pipeline starts.

```bash
docker exec sqlflow-postgres psql -U postgres -c "
SELECT bucket, lang, posts
FROM posts_per_minute_by_lang
WHERE bucket = (SELECT max(bucket) FROM posts_per_minute_by_lang)
ORDER BY posts DESC LIMIT 8;"
```

```text
         bucket         |  lang   | posts
------------------------+---------+-------
 2026-09-15 00:51:00+00 | en      |  1720
 2026-09-15 00:51:00+00 | unknown |   234
 2026-09-15 00:51:00+00 | ja      |   180
 2026-09-15 00:51:00+00 | pt      |   102
 2026-09-15 00:51:00+00 | es      |    43
 2026-09-15 00:51:00+00 | ne      |    32
 2026-09-15 00:51:00+00 | ko      |    30
 2026-09-15 00:51:00+00 | ja-JP   |    10
(8 rows)
```

--------------------------------

### Create telemetry schema, devices table, and readings hypertable

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Run this SQL in psql or the console's SQL editor to set up the telemetry schema, the devices reference table with seed data, and the readings hypertable. The readings table uses a primary key on (time, device_id) so the SQLFlow sink's ON CONFLICT DO NOTHING can safely deduplicate at-least-once deliveries. The site and model columns are nullable so readings from unknown devices are still accepted.

```sql
CREATE SCHEMA telemetry;
SET search_path TO telemetry;

CREATE TABLE devices (
  device_id TEXT PRIMARY KEY,
  site      TEXT NOT NULL,
  model     TEXT NOT NULL
);

INSERT INTO devices (device_id, site, model) VALUES
  ('d-100', 'oslo',   'tx-2'),
  ('d-200', 'lisbon', 'tx-2'),
  ('d-300', 'lisbon', 'tx-3');

CREATE TABLE readings (
  time      TIMESTAMPTZ      NOT NULL,
  device_id TEXT             NOT NULL,
  site      TEXT,
  model     TEXT,
  value     DOUBLE PRECISION,
  PRIMARY KEY (time, device_id)
) WITH (
  timescaledb.hypertable,
  timescaledb.chunk_interval = '7 days'
);
```

--------------------------------

### Install Python dependencies

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Installs the repository's Python dependencies, which are used to publish sample data. SQLFlow itself does not need Python.

```bash
uv sync
```

--------------------------------

### Run SQLFlow pipeline

Source: https://turbolytics.io/docs/sqlflow/tutorials/clickhouse-sink

Runs SQLFlow with the example config, reading from Kafka and writing to ClickHouse in batches of 50k. The output shows throughput logs; three consecutive runs on an Apple M1 Pro measured 104,605, 107,014 and 105,742 messages per second, all writing 1,000,000 rows.

```bash
./bin/sqlflow run $(pwd)/dev/config/examples/kafka.clickhouse.yml --max-msgs=1000000
```

```text
INFO  consumer loop starting
INFO  throughput  {"messages_consumed": 500000, "total_throughput_per_second": 100023}
INFO  max messages consumed, stopping consumer loop
INFO  throughput  {"messages_consumed": 1000000, "total_throughput_per_second": 104605}
```

--------------------------------

### Full SQLFlow configuration (kafka.ducklake.yml)

Source: https://turbolytics.io/docs/sqlflow/tutorials/ducklake-sink

Complete example configuration for a SQLFlow pipeline that reads from Kafka, transforms data with SQL, and writes to DuckLake. Includes commands to install/attach DuckLake and create the target table, with a batch size template default of 1.

```yaml
commands:
  - name: install ducklake
    sql: |
      INSTALL ducklake;
      LOAD ducklake;

  - name: attach to ducklake
    sql: |
      ATTACH 'ducklake:my_ducklake.ducklake' AS my_ducklake;
      USE my_ducklake;

  - name: create events table if not exists
    sql: |
      CREATE TABLE IF NOT EXISTS my_ducklake.events (
        ip TEXT,
        event TEXT,
        properties_city TEXT,
        properties_country TEXT,
        timestamp TIMESTAMP,
        type TEXT,
        userId TEXT
      );

pipeline:
  batch_size: {{ SQLFLOW_BATCH_SIZE|default(1) }}
  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: motherduck-sink-1
      auto_offset_reset: earliest
      topics:
        - {{ SQLFLOW_INPUT_TOPIC|default('input-user-clicks-ducklake') }}

  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      INSERT INTO my_ducklake.events
      SELECT
        ip,
        event,
        properties ->> 'city' AS properties_city,
        properties ->> 'country' AS properties_country,
        CAST(timestamp AS TIMESTAMP) AS timestamp,
        type,
        userId
      FROM batch;

  sink:
    type: noop
```

--------------------------------

### Example JSON message for batch table

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Example JSON message that produces a batch table with inferred schema. Nested objects become DuckDB structs and arrays become lists.

```json
{"user_id":7,"ts":"2026-09-07T10:00:00","props":{"city":"NYC"},"tags":["a","b"]}
```

--------------------------------

### Build SQLFlow binary with make sqlflow

Source: https://turbolytics.io/docs/sqlflow/introduction/local-development

Clones the SQLFlow repository and builds the binary. The binary lands at bin/sqlflow. To build a released version, check out its tag first; v1.0.6 is the current one. If you don't have a Go toolchain, build the container instead with make sqlflow-image.

```bash
git clone https://github.com/turbolytics/sql-flow.git
cd sql-flow

make sqlflow
make setup-dev
```

--------------------------------

### Postgres sink encode failure error output

Source: https://turbolytics.io/docs/sqlflow/operations/handling-errors

Example error log when the Postgres sink cannot encode a value for a column. This is a user error (`user.sink.encode_failed`) and exits 10 on the first attempt. The example shows a BIGINT value sent to a smallint column, with SQLSTATE 57014.

```text
ERROR  error flushing sink  {"error": "[user.sink.encode_failed] postgres sink: encode a value for its column (SQLSTATE 57014): ERROR: COPY from stdin failed: unable to encode 1099511627776 into binary format for int2 (OID 21): 1099511627776 is greater than maximum value for int2 (SQLSTATE 57014)", "error.code": "user.sink.encode_failed", "error.class": "user"}
```

--------------------------------

### Publish test data to Kafka

Source: https://turbolytics.io/docs/sqlflow/tutorials/motherduck-sink

Runs the sample-data script to send one million mock user interaction events into Kafka for SQLFlow to ingest.

```bash
uv run python3 cmd/publish-test-data.py --num-messages=1000000 --topic="input-user-clicks-motherduck"
```

--------------------------------

### Publish Kafka Message

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Starts the Kafka console producer and publishes a JSON message to the input topic.

```bash
docker exec -it kafka1 kafka-console-producer --bootstrap-server=kafka1:9092 --topic=input-kafka-postgres-sink
>
```

```json
>{"timestamp": "2025-01-18T12:34:56Z", "user_id": "u12345", "action": "button_click", "browser": "Chrome"}
```

--------------------------------

### DLQ error message payload

Source: https://turbolytics.io/docs/sqlflow/operations/handling-errors

Example structured error message published to the DLQ. Includes the failing payload, the phase, and the coded error.

```json
{
  "error": "[user.data.malformed] invalid json: \"!invalid!\"",
  "message": "!invalid!",
  "phase": "handler.write",
  "timestamp": "2026-09-06T18:28:31.015Z"
}
```

--------------------------------

### SQLFlow startup log output

Source: https://turbolytics.io/docs/sqlflow/tutorials/tigerdata-enrich

Expected log output showing the command steps and Kafka consumer initialization. The pipeline is idle, waiting for messages.

```text
INFO  Executing command step  {"name": "load the postgres extension"}
INFO  Executing command step  {"name": "attach the service"}
INFO  initializing kafka source  {"topics": "[readings]", "group.id": "tigerdata-enrich", "auto.offset.reset": "earliest"}
INFO  consumer loop starting
```

--------------------------------

### Print SQLFlow configuration reference

Source: https://turbolytics.io/docs/sqlflow/introduction/configuration

Prints every configuration key with its type and a comment, generated from the schema the validator uses.

```bash
./bin/sqlflow config example
```

--------------------------------

### Start Postgres container

Source: https://turbolytics.io/docs/sqlflow/tutorials/tumbling-window-postgres

Restarts the Postgres container after an outage. A supervisor restarts the process once Postgres is back; the recovery depends on where the pipeline state lives.

```bash
docker start sqlflow-postgres
```

--------------------------------

### Path normalization example

Source: https://turbolytics.io/docs/sqlflow/tutorials/log-reduction

Shows how raw paths are normalized into route classes: the filename is dropped, two directory segments are kept, digit runs are folded to 'N', and the extension classifies the route. This collapses 21,214 distinct paths into 3,423 classes.

```text
/shuttle/missions/sts-73/mission-sts-73.html   ->  /shuttle/missions/*.html
/images/NASA-logosmall.gif                     ->  /images/*.gif
```

--------------------------------

### Start SQLFlow with batch size 200

Source: https://turbolytics.io/docs/sqlflow/tutorials/postgres-sink

Run SQLFlow in a Docker container with SQLFLOW_BATCH_SIZE=200, pointing to Postgres and Kafka. This is the second step; it consumes the 200k messages and inserts them into the user_action table. The command uses the turbolytics/sql-flow:v2026.09.14.1 image and the kafka.postgres.sink.yml config.

```bash
docker run -v $(pwd)/dev:/tmp/conf \
  -e SQLFLOW_BATCH_SIZE=200 \
  -e SQLFLOW_POSTGRES_USERS_URI=postgresql://postgres:postgres@host.docker.internal:5432/testdb \
  -e SQLFLOW_KAFKA_BROKERS=host.docker.internal:29092 \
  turbolytics/sql-flow:v2026.09.14.1 \
  run /tmp/conf/config/examples/kafka.postgres.sink.yml --max-msgs=200000
```

--------------------------------

### Sample output of Bluesky firehose messages

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

Example of the raw Bluesky firehose messages that the Kafka consumer displays. This output shows the structure of the JSON payloads.

```json
...
{"commit":{"cid":"bafyreibrgt7sv6opy2jyd3mxfgxhu7xzyaciv77xba7q33ukjzloghe4r4","collection":"app.bsky.feed.post","operation":"create","record":{"$type":"app.bsky.feed.post","createdAt":"2026-09-13T20:40:44.645Z","langs":["en"],"reply":{"parent":{"cid":"bafyreiggjkiiz7m2n5sa5dduq3soeikvrri6yqlhcrfcjrzy6trewwd27q","uri":"at://did:plc:hjdxvrq7rh43z5umtt3mucxw/app.bsky.feed.post/3mvge7lmfrk2o"},"root":{"cid":"bafyreiggjkiiz7m2n5sa5dduq3soeikvrri6yqlhcrfcjrzy6trewwd27q","uri":"at://did:plc:hjdxvrq7rh43z5umtt3mucxw/app.bsky.feed.post/3mvge7lmfrk2o"}},"text":"The urgent care is important if you have more than 3 meximelts"},"rev":"3mvghejkr5c24","rkey":"3mvgheik5m22k"},"did":"did:plc:hvnzm4bp66p4b2lqelfwtlwz","kind":"commit","time_us":1789332046760134}
{"commit":{"cid":"bafyreiadonzloj2mudhnolm7lggqa2hyqy5mwvlq4zmq3em42cncm4y3km","collection":"app.bsky.feed.post","operation":"create","record":{"$type":"app.bsky.feed.post","createdAt":"2026-09-13T20:40:44.498Z","embed":{"$type":"app.bsky.embed.external","external":{"description":"YouTube video by Clash With Bao","thumb":{"$type":"blob","mimeType":"image/jpeg","ref":{"$link":"bafkreibvghszoauw2kbtoz5sqkh5on5dzh2tjxeub5rnxlnuipya3ira6u"},"size":82250},"title":"4th attempt at \"Arraignment\" for a 3rd degree \"FELONY\" By the Bexar County Clowns #cops, #court","uri":"https://youtube.com/shorts/XCMTnI9X8RU?is=kQtqH9Zc9uzr1h3k"}},"langs":["en"],"text":"Legendary Clash with Bao."},"rev":"3mvgheja5up2n","rkey":"3mvgheifl4k2d"},"did":"did:plc:ul4zc6gkcsupdwhwqpucipam","kind":"commit","time_us":1789332046784256}
...
```

--------------------------------

### Transform Stream and Write to Stdout: SQL Query

Source: https://turbolytics.io/docs/sqlflow/tutorials/bluesky-firehose

SQL transformation that extracts the timestamp and post kind from the raw event. Used in the configuration file for the stdout example.

```sql
SELECT
  strftime(to_timestamp(time_us / 1000000), '%Y-%m-%dT%H:%M:%S') as time,
  kind
FROM batch
```
Tokens:
Raw