SQLFlow and ClickHouse

clickhouses://default:<password>@<host>.clickhouse.cloud:8443/<database>

sink:
  type: clickhouse
  clickhouse:
    dsn: {{ SQLFLOW_CLICKHOUSE_DSN }}
    table: trips_small

export SQLFLOW_CLICKHOUSE_DSN='clickhouses://default:<password>@<host>.clickhouse.cloud:8443/nyc_taxi'

export SQLFLOW_CLICKHOUSE_DSN='clickhouse://default:<password>@clickhouse.internal:8123/nyc_taxi'

CREATE DATABASE IF NOT EXISTS nyc_taxi;

CREATE TABLE IF NOT EXISTS nyc_taxi.trips_small (
    trip_id             UInt32,
    pickup_datetime     DateTime,
    dropoff_datetime    DateTime,
    pickup_longitude    Nullable(Float64),
    pickup_latitude     Nullable(Float64),
    dropoff_longitude   Nullable(Float64),
    dropoff_latitude    Nullable(Float64),
    passenger_count     UInt8,
    trip_distance       Float32,
    fare_amount         Float32,
    extra               Float32,
    tip_amount          Float32,
    tolls_amount        Float32,
    total_amount        Float32,
    payment_type        Enum('CSH' = 1, 'CRE' = 2, 'NOC' = 3, 'DIS' = 4, 'UNK' = 5),
    pickup_ntaname      LowCardinality(String),
    dropoff_ntaname     LowCardinality(String)
)
ENGINE = MergeTree
PRIMARY KEY (pickup_datetime, dropoff_datetime);

pip install confluent-kafka
curl -O https://datasets-documentation.s3.eu-west-3.amazonaws.com/nyc-taxi/trips_0.gz

kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic nyc-taxi-trips --partitions 1 --replication-factor 1

import gzip, json
from confluent_kafka import Producer

COLUMNS = ["trip_id", "pickup_datetime", "dropoff_datetime", "pickup_longitude",
           "pickup_latitude", "dropoff_longitude", "dropoff_latitude", "passenger_count",
           "trip_distance", "fare_amount", "extra", "tip_amount", "tolls_amount",
           "total_amount", "payment_type", "pickup_ntaname", "dropoff_ntaname"]
INTS = {"trip_id", "passenger_count"}
FLOATS = {"pickup_longitude", "pickup_latitude", "dropoff_longitude", "dropoff_latitude",
          "trip_distance", "fare_amount", "extra", "tip_amount", "tolls_amount", "total_amount"}

producer = Producer({"bootstrap.servers": "localhost:9092"})

def send(topic, value):
    # The producer's queue holds 100,000 messages by default, and this file
    # has a million rows; poll() drains it as it fills.
    while True:
        try:
            producer.produce(topic, value)
            return
        except BufferError:
            producer.poll(0.5)

with gzip.open("trips_0.gz", "rb") as f:
    header = f.readline().decode().rstrip("\n").split("\t")
    index = {name: i for i, name in enumerate(header)}
    for raw in f:
        # Split on the raw bytes and decode only the columns being published.
        # The file carries columns this example does not use that are not
        # valid UTF-8, and decoding the whole line would have to corrupt them.
        parts = raw.rstrip(b"\n").split(b"\t")
        row = {}
        for col in COLUMNS:
            value = parts[index[col]].decode()
            if value == "":
                row[col] = None
            elif col in INTS:
                row[col] = int(float(value))
            elif col in FLOATS:
                row[col] = float(value)
            else:
                row[col] = value
        send("nyc-taxi-trips", json.dumps(row))
producer.flush()

python publish.py

pipeline:
  name: nyc-taxi-clickhouse
  batch_size: 20000

  source:
    type: kafka
    kafka:
      brokers: [{{ SQLFLOW_KAFKA_BROKERS|default('localhost:9092') }}]
      group_id: nyc-taxi
      auto_offset_reset: earliest
      topics:
        - nyc-taxi-trips

  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      SELECT
        CAST(trip_id AS UINTEGER)            AS trip_id,
        CAST(pickup_datetime AS TIMESTAMP)   AS pickup_datetime,
        CAST(dropoff_datetime AS TIMESTAMP)  AS dropoff_datetime,
        pickup_longitude,
        pickup_latitude,
        dropoff_longitude,
        dropoff_latitude,
        CAST(passenger_count AS UTINYINT)    AS passenger_count,
        CAST(trip_distance AS FLOAT)         AS trip_distance,
        CAST(fare_amount AS FLOAT)           AS fare_amount,
        CAST(extra AS FLOAT)                 AS extra,
        CAST(tip_amount AS FLOAT)            AS tip_amount,
        CAST(tolls_amount AS FLOAT)          AS tolls_amount,
        CAST(total_amount AS FLOAT)          AS total_amount,
        payment_type,
        pickup_ntaname,
        dropoff_ntaname
      FROM batch

  sink:
    type: clickhouse
    clickhouse:
      dsn: {{ SQLFLOW_CLICKHOUSE_DSN }}
      table: trips_small

docker run --rm \
  --network <the broker's docker network> \
  -v "$(pwd)":/conf \
  -e SQLFLOW_CLICKHOUSE_DSN \
  -e SQLFLOW_KAFKA_BROKERS=<broker address as advertised inside that network> \
  turbolytics/sql-flow:v2026.10.08 \
  run /conf/taxi.yml --max-msgs=1000000

SELECT payment_type, count() AS trips, round(avg(total_amount), 2) AS avg_total
FROM nyc_taxi.trips_small
GROUP BY payment_type
ORDER BY trips DESC;

┌─payment_type─┬──trips─┬─avg_total─┐
│ CSH          │ 617237 │     17.82 │
│ CRE          │ 377897 │     13.91 │
│ NOC          │   3573 │     13.89 │
│ DIS          │   1293 │     12.09 │
└──────────────┴────────┴───────────┘

CREATE TABLE bluesky_posts (
    ts    DateTime,
    did   String,
    langs Array(String),
    text  String
) ENGINE = MergeTree ORDER BY ts;

pipeline:
  batch_size: 200
  source:
    type: websocket
    websocket:
      uri: 'wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post'
  handler:
    type: 'handlers.InferredMemBatch'
    sql: |
      SELECT
        to_timestamp(time_us / 1000000)          AS ts,
        did,
        COALESCE(commit.record.langs, [])       AS langs,
        COALESCE(commit.record.text, '')        AS text
      FROM batch
  sink:
    type: clickhouse
    clickhouse:
      dsn: {{ SQLFLOW_CLICKHOUSE_DSN }}
      table: bluesky_posts

docker run --rm \
  -v "$(pwd)":/conf \
  -e SQLFLOW_CLICKHOUSE_DSN \
  turbolytics/sql-flow:v2026.10.08 \
  run /conf/bluesky.yml --max-msgs=2000

sqlflow run bluesky.yml --max-msgs=2000

SELECT lang, count() AS posts
FROM (SELECT arrayJoin(langs) AS lang FROM bluesky_posts)
GROUP BY lang ORDER BY posts DESC LIMIT 5;