Skip to main content
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 without credentials.

​
Prerequisites

1

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}
2

Create a basin

export BASIN="YOUR_BASIN_NAME"
s2 create-basin ${BASIN}
3

Install Streamling

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

Install the community plugins

Download the shared library for your platform from the releases page and tell Streamling where to find it:
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
export STREAMLING__PLUGIN__PATH="$PWD/libcommunity_plugins.dylib"
Use the darwin-x86_64 asset on Intel Macs.

​
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.
1

Create event streams and append some events

s2 create-stream s2://${BASIN}/events-web
s2 create-stream s2://${BASIN}/events-mobile
Append an event to each stream:
curl -s -X POST "https://${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 -s -X POST "https://${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}}
2

Start ClickHouse

docker run -d --name clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server
3

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.
4

Run it

Connection details are passed as environment variables, keeping secrets out of the pipeline definition:
STREAMLING__PLUGIN__S2_SOURCE__BASIN="${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.
5

Query ClickHouse

In another terminal:
docker exec 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.

​
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.
1

Start Postgres with logical replication

docker run -d --name postgres-cdc -p 5433:5432 \
  -e POSTGRES_PASSWORD=postgres postgres:16 -c wal_level=logical
Create a table with some rows:
docker exec postgres-cdc psql -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);"
2

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.
3

Run it

STREAMLING__PLUGIN__S2_SINK__BASIN="${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.
4

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://${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 and make some changes:
docker exec postgres-cdc psql -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 clickhouse -p 8123:8123 \
  -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server
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__S2_SOURCE__BASIN="${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. Querying ClickHouse shows the update applied and the deleted row gone: 
docker exec 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, and replays apply cleanly since sinks upsert by primary key.