> ## Documentation Index > Fetch the complete documentation index at: https://s2.dev/docs/llms.txt > Use this file to discover all available pages before exploring further. # Streamling > Use Streamling's S2 source and sink plugins to connect S2 with Streamling pipelines. [Streamling](https://github.com/goldsky-io/streamling) is an open source columnar streaming runtime from [Goldsky](https://goldsky.com/), built for high-throughput real-time pipelines with checkpointing and at-least-once delivery. S2 support ships as two plugins in [streamling-community-plugins](https://github.com/goldsky-io/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](https://debezium.io/)-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](https://github.com/goldsky-io/streamling-community-plugins#available-plugins). Both plugins accept an `endpoint` option, so pipelines can run against [s2-lite](/s2-lite) — handy for CI and local development without credentials. ## Prerequisites Generate an access token from the [dashboard](https://s2.dev/dashboard), then: ```bash theme={null} export S2_ACCESS_TOKEN="YOUR_ACCESS_TOKEN" ``` Install the [S2 CLI](/quickstart#get-started-with-the-cli) and point it at the same token: ```bash theme={null} s2 config set access_token ${S2_ACCESS_TOKEN} ``` ```bash theme={null} export BASIN="YOUR_BASIN_NAME" s2 create-basin ${BASIN} ``` ```bash theme={null} curl -fsSL https://www.streamling.dev/install.sh | bash ``` Download the shared library for your platform from the [releases page](https://github.com/goldsky-io/streamling-community-plugins/releases) and tell Streamling where to find it: ```bash theme={null} 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. ```bash theme={null} 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 export STREAMLING__PLUGIN__PATH="$PWD/libcommunity_plugins.so" ``` Use the `linux-aarch64` asset on ARM. ## Example: Events into ClickHouse This example sends `POST → S2 → Streamling → ClickHouse`: services append JSON events to S2 streams over the [REST API](/api/protocol), and a Streamling pipeline tails every stream under a prefix, decodes the events into typed columns, and upserts them into ClickHouse. ```bash theme={null} s2 create-stream s2://${BASIN}/events-web s2 create-stream s2://${BASIN}/events-mobile ``` Append an event to each stream: ```bash theme={null} 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: ```json theme={null} {"start":{"seq_num":0,"timestamp":1783548986349},"end":{"seq_num":1,"timestamp":1783548986349},"tail":{"seq_num":1,"timestamp":1783548986349}} ``` ```bash theme={null} docker run -d --name clickhouse -p 8123:8123 \ -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server ``` Create `events-to-clickhouse.yaml`: ```yaml theme={null} 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. Connection details are passed as environment variables, keeping secrets out of the pipeline definition: ```bash theme={null} 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. In another terminal: ```bash theme={null} 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. ```bash theme={null} 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: ```bash theme={null} 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);" ``` Create `postgres-cdc-to-s2.yaml`: ```yaml theme={null} 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](https://github.com/goldsky-io/streamling-community-plugins#postgres-cdc-source-postgres_cdc_source) for production guidance, including least-privilege Postgres permissions. ```bash theme={null} 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. In another terminal, the initial table copy is already there, each record carrying its operation as a header: ```bash theme={null} s2 read s2://${BASIN}/cdc/products -s 0 --format json ``` ```json theme={null} {"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: ```bash theme={null} 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: ```json theme={null} {"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: ```bash theme={null} docker run -d --name clickhouse -p 8123:8123 \ -e CLICKHOUSE_PASSWORD=password clickhouse/clickhouse-server ``` In a new terminal, create `cdc-to-clickhouse.yaml`: ```yaml theme={null} 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: ```bash theme={null} 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: ```bash theme={null} 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](https://github.com/goldsky-io/streamling#state-backends) for production. Delivery is at-least-once end to end, and replays apply cleanly since sinks upsert by primary key.