Streamling

Streamling is an open source columnar streaming runtime from Goldsky, built for high-throughput real-time pipelines with checkpointing and at-least-once delivery. S2 support ships as two plugins in streamling-community-plugins v0.2.0 and later:

s2_source reads S2 streams into Arrow batches for any downstream Streamling transform or sink — exact stream names or a prefix, with newly created streams discovered automatically. Records can flow through as the raw envelope (stream, seq_num, timestamp, headers, body) or as typed columns decoded from JSON bodies.
s2_sink appends rows from any Streamling source or transform as JSON records to an S2 stream, with optional per-row routing across streams via a template like events/{tenant}.

Use s2_source when S2 is the durable input to a Streamling pipeline, and s2_sink when S2 is the durable output. The examples below use ClickHouse and Postgres CDC to show useful Streamling sources and sinks in both directions, but the S2 plugins compose with Streamling’s other sources, sinks, and transforms. The two plugins also compose into CDC round trips: the sink carries each row’s operation as a Debezium-style dbz.op record header, and the source restores it, so inserts, updates, and deletes survive a trip through S2 and apply correctly in a downstream database.

Reference

Configuration tables for both plugins are in the streamling-community-plugins README.

Both plugins accept an endpoint option, so pipelines can run against s2-lite — handy for CI and local development. See the CLI endpoint overrides for the accompanying CLI setup; use any non-empty access token value locally, since s2-lite ignores it.

Prerequisites

Set up the S2 CLI

Generate an access token from the dashboard, then:

export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN"

Install the S2 CLI and point it at the same token:

s2 config set access_token ${S2_ACCESS_TOKEN}

Create a basin

export S2_BASIN="YOUR_BASIN_NAME"
s2 create-basin ${S2_BASIN}

Install and start Docker

The ClickHouse and Postgres examples use Docker. Install Docker, start it, and verify that the daemon is available:

docker info

Install Streamling

curl -fsSL https://www.streamling.dev/install.sh | bash

Create a walkthrough directory

Keep the pipeline files and checkpoint state for this walkthrough together:

mkdir -p "$HOME/streamling-s2-demo"
cd "$HOME/streamling-s2-demo"

Install the community plugins

Download the shared library for your platform from the releases page and tell Streamling where to find it:

macOS
Linux

mkdir -p "$HOME/.streamling/plugins"
curl -fsSLO https://github.com/goldsky-io/streamling-community-plugins/releases/latest/download/community-plugins-darwin-aarch64.tar.gz
tar -xzf community-plugins-darwin-aarch64.tar.gz -C "$HOME/.streamling/plugins"

Use the darwin-x86_64 asset on Intel Macs.

mkdir -p "$HOME/.streamling/plugins"
curl -fsSLO https://github.com/goldsky-io/streamling-community-plugins/releases/latest/download/community-plugins-linux-x86_64.tar.gz
tar -xzf community-plugins-linux-x86_64.tar.gz -C "$HOME/.streamling/plugins"

Use the linux-aarch64 asset on ARM.

The run commands below set STREAMLING__PLUGIN__PATH inline, so they also work in a fresh terminal.

Run every example from $HOME/streamling-s2-demo. Streamling stores checkpoints there in state.db; see Clean up and rerun before replaying the walkthrough from the beginning.

Shell variables and the working directory do not carry into a new terminal. Whenever the walkthrough opens another terminal, return to the walkthrough directory and set S2_BASIN again; also set S2_ACCESS_TOKEN before starting a Streamling pipeline:

cd "$HOME/streamling-s2-demo"
export S2_BASIN="YOUR_BASIN_NAME"
export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN"

Example: Events into ClickHouse

This example sends POST → S2 → Streamling → ClickHouse: services append JSON events to S2 streams over the REST API, and a Streamling pipeline tails every stream under a prefix, decodes the events into typed columns, and upserts them into ClickHouse.

Create event streams and append some events

s2 create-stream s2://${S2_BASIN}/events-web
s2 create-stream s2://${S2_BASIN}/events-mobile

Append an event to each stream:

curl -fsS -X POST "https://${S2_BASIN}.b.s2.dev/v1/streams/events-web/records" \
  -H "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"records": [{"body": "{\"id\":1,\"event\":\"page_view\",\"amount\":0.25}"}]}'

curl -fsS -X POST "https://${S2_BASIN}.b.s2.dev/v1/streams/events-mobile/records" \
  -H "Authorization: Bearer ${S2_ACCESS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{"records": [{"body": "{\"id\":2,\"event\":\"app_open\",\"amount\":0.75}"}]}'

Each append is acknowledged with the durable sequence number range:

{"start":{"seq_num":0,"timestamp":1783548986349},"end":{"seq_num":1,"timestamp":1783548986349},"tail":{"seq_num":1,"timestamp":1783548986349}}

Start ClickHouse

docker run -d --name s2-streamling-demo-clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server &&
curl -fs --connect-timeout 2 --retry 30 --retry-delay 1 \
  --retry-connrefused --retry-max-time 60 http://localhost:8123/ping > /dev/null ||
  { echo "ClickHouse did not become ready" >&2; false; }

Write the pipeline

Create events-to-clickhouse.yaml:

sources:
  events:
    type: s2_source
    stream_prefix: events-
    schema: "id:int64,event:string,amount:float64"
    include_metadata: true
    primary_key: id

transforms: {}

sinks:
  ch_events:
    type: clickhouse
    from: events
    table: events
    primary_key: id

The schema option decodes JSON bodies into typed columns, and include_metadata adds _s2_stream, _s2_seq_num, and _s2_timestamp columns. The ClickHouse sink creates the table on first write.

Run it

Connection details are passed as environment variables, keeping secrets out of the pipeline definition:

STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SOURCE__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SOURCE__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
STREAMLING__CLICKHOUSE_SINK__PASSWORD=password \
streamling events-to-clickhouse.yaml

The pipeline reads both streams from the beginning and keeps tailing them for new records. Leave it running while you query ClickHouse in the next step.

Query ClickHouse

In another terminal:

docker exec s2-streamling-demo-clickhouse clickhouse-client --password password \
  --query "SELECT id, event, amount, _s2_stream FROM events FINAL ORDER BY id"

1	page_view	0.25	events-web
2	app_open	0.75	events-mobile

Append more events with curl and they show up in ClickHouse within moments. You can even create a brand new events-* stream: the source re-lists the prefix every 60 seconds (tunable via update_streams_interval_secs) and starts reading it automatically.

Events posted to S2 are flowing into ClickHouse as typed rows.

When you are ready to continue, return to the terminal running this pipeline and press Ctrl+C.

Example: Postgres CDC into S2

This example sends Postgres CDC → Streamling → S2: Streamling’s postgres_cdc_source performs an initial table copy followed by continuous logical replication, and s2_sink appends every change to an S2 stream with the operation preserved as a dbz.op record header.

Start Postgres with logical replication

docker run -d --name s2-streamling-demo-postgres -p 5433:5432 \
  -e POSTGRES_PASSWORD=postgres postgres:16 -c wal_level=logical &&
for attempt in {1..30}; do
  if docker exec s2-streamling-demo-postgres pg_isready -h 127.0.0.1 -U postgres > /dev/null 2>&1; then
    break
  elif [ "${attempt}" -eq 30 ]; then
    echo "Postgres did not become ready" >&2
    false
  else
    sleep 1
  fi
done

Create a table with some rows:

docker exec s2-streamling-demo-postgres psql -v ON_ERROR_STOP=1 -U postgres -c "
  CREATE TABLE products (id BIGINT PRIMARY KEY, name TEXT, price DOUBLE PRECISION);
  INSERT INTO products VALUES (1, 'Avocados', 5.99), (2, 'Salmon', 14.99), (3, 'Sourdough', 6.99);"

Write the pipeline

Create postgres-cdc-to-s2.yaml:

sources:
  products:
    type: postgres_cdc_source
    host: localhost
    port: 5433
    database: postgres
    username: postgres
    password: postgres
    table: products
    primary_key: id
    publication_name: streamling_products
    slot_name: streamling_products
    memory_backpressure_enabled: false

transforms: {}

sinks:
  s2_products:
    type: s2_sink
    from: products
    stream: cdc/products

The target stream is created automatically on first append (ensure_stream defaults to true).

memory_backpressure_enabled: false suits local machines, where system memory use often sits above the plugin’s default backpressure threshold. See the plugin README for production guidance, including least-privilege Postgres permissions.

Run it

STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SINK__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SINK__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
streamling postgres-cdc-to-s2.yaml

Leave this pipeline running while you read from S2 and apply Postgres changes.

Read the changes from S2

In another terminal, the initial table copy is already there, each record carrying its operation as a header:

s2 read s2://${S2_BASIN}/cdc/products -s 0 --format json

{"seq_num":0,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":1,\"name\":\"Avocados\",\"price\":5.99}"}
{"seq_num":1,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":2,\"name\":\"Salmon\",\"price\":14.99}"}
{"seq_num":2,"timestamp":1783549032082,"headers":[["dbz.op","c"]],"body":"{\"id\":3,\"name\":\"Sourdough\",\"price\":6.99}"}

Leave the read session tailing. In a third terminal, return to the walkthrough directory, set S2_BASIN, and make some changes:

docker exec s2-streamling-demo-postgres psql -v ON_ERROR_STOP=1 -U postgres -c "
  UPDATE products SET price = 4.99 WHERE id = 1;
  DELETE FROM products WHERE id = 2;"

The update and delete stream in with u and d ops:

{"seq_num":3,"timestamp":1783549052191,"headers":[["dbz.op","u"]],"body":"{\"id\":1,\"name\":\"Avocados\",\"price\":4.99}"}
{"seq_num":4,"timestamp":1783549052191,"headers":[["dbz.op","d"]],"body":"{\"id\":2}"}

Postgres changes are flowing into your S2 stream in order, with CDC semantics intact.

Example: Postgres to ClickHouse through S2

Because s2_source maps dbz.op headers back to row operations, the CDC stream from the previous section can drive a second pipeline into ClickHouse — S2 acting as the durable, replayable changelog between the two. If the ClickHouse container from the first example is not already running, start it:

docker run -d --name s2-streamling-demo-clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server &&
curl -fs --connect-timeout 2 --retry 30 --retry-delay 1 \
  --retry-connrefused --retry-max-time 60 http://localhost:8123/ping > /dev/null ||
  { echo "ClickHouse did not become ready" >&2; false; }

In a new terminal, create cdc-to-clickhouse.yaml:

sources:
  products_cdc:
    type: s2_source
    streams: cdc/products
    schema: "id:int64,name:string?,price:float64?"
    primary_key: id

transforms: {}

sinks:
  ch_products:
    type: clickhouse
    from: products_cdc
    table: products
    primary_key: id

Run the second pipeline:

STREAMLING__PLUGIN__PATH="$HOME/.streamling/plugins" \
STREAMLING__PLUGIN__S2_SOURCE__BASIN="${S2_BASIN}" \
STREAMLING__PLUGIN__S2_SOURCE__ACCESS_TOKEN="${S2_ACCESS_TOKEN}" \
STREAMLING__CLICKHOUSE_SINK__PASSWORD=password \
streamling cdc-to-clickhouse.yaml

Non-key columns are marked nullable (?) since delete records carry only the primary key. In another terminal, query ClickHouse to see the update applied and the deleted row gone:

docker exec s2-streamling-demo-clickhouse clickhouse-client --password password \
  --query "SELECT id, name, price FROM products FINAL ORDER BY id"

1	Avocados	4.99
3	Sourdough	6.99

Streamling checkpoints source positions in its state backend — by default a SQLite file at ./state.db, so a restarted pipeline resumes where it left off rather than re-reading streams. Postgres is available as a state backend for production. Delivery is at-least-once end to end. The ClickHouse sink converges on the latest row per primary key, so replayed rows do not change the final table state. The S2 changelog itself can contain duplicates after ambiguous retries or checkpoint replay; consumers should deduplicate if each change must be processed exactly once.

Clean up and rerun

Stop every Streamling process and the tailing s2 read command with Ctrl+C. Then remove the walkthrough’s containers and their anonymous data volumes:

docker rm -fv s2-streamling-demo-clickhouse s2-streamling-demo-postgres

To replay the walkthrough from sequence number 0, also remove its checkpoint file and streams. The S2_BASIN guard prevents accidental deletion from an unset variable; only continue if the named streams were created for this walkthrough:

if [ -z "${S2_BASIN:-}" ]; then
  echo "Set S2_BASIN to the walkthrough basin before cleaning up" >&2
else
  rm -f "$HOME/streamling-s2-demo/state.db"
  s2 delete-stream s2://${S2_BASIN}/events-web
  s2 delete-stream s2://${S2_BASIN}/events-mobile
  s2 delete-stream s2://${S2_BASIN}/cdc/products
fi

Wait for the stream deletions to finish before recreating them. Keep the existing basin and resume at Events into ClickHouse instead of creating the basin again.

AI SDKs

Observability

Pipelines

Sync

Streamling

Reference

Prerequisites

Example: Events into ClickHouse

Example: Postgres CDC into S2

Example: Postgres to ClickHouse through S2

Clean up and rerun

​Reference

​Prerequisites

​Example: Events into ClickHouse

​Example: Postgres CDC into S2

​Example: Postgres to ClickHouse through S2

​Clean up and rerun

Reference

Prerequisites

Example: Events into ClickHouse

Example: Postgres CDC into S2

Example: Postgres to ClickHouse through S2

Clean up and rerun