Substreams: Run a sink

Build a queryable transfer history in Postgres from a Substreams package. You will run the SQL sink locally, stream a bounded Ethereum range through GraphOps, and verify the stored rows.

The SQL sink is part of the Substreams CLI as substreams sink postgres and substreams sink clickhouse; there is no separate binary to install. The example uses substreams-template@v0.3.1, a public Ethereum package with a db_out module that emits sf.substreams.sink.database.v1.DatabaseChanges and a SQL sink configuration.

You run the sink in this guide. For GraphOps-hosted sinks, contact support about enterprise options.

The package runs on GraphOps. You run setup once, run the sink over a bounded range, then query the rows it wrote. The sink stores its cursor in the same database.

Prerequisites

A Substreams API key.
The Substreams CLI. Install it with brew install streamingfast/tap/substreams, or follow Install Substreams CLI.
Docker, or an existing Postgres database.
A package with SQL sink output. This guide uses substreams-template@v0.3.1.

Start Postgres

Start a local Postgres instance for the sink. The sample credentials are for local development only; use managed secrets for production deployments.

docker run --name graphops-substreams-postgres \
  -e POSTGRES_USER=dev-node \
  -e POSTGRES_PASSWORD=local-dev-only-password \
  -e POSTGRES_DB=dev-node \
  -p 5432:5432 \
  -d postgres:16

Set the connection string (DSN) for the sink and for psql:

export DSN="postgres://dev-node:local-dev-only-password@localhost:5432/dev-node?sslmode=disable"

The sink takes the DSN from --dsn, or from the SUBSTREAMS_SINK_DSN environment variable when you omit the flag. See the DSN reference for Postgres and ClickHouse options.

Configure GraphOps access

Set your API key. The sink reads it from SUBSTREAMS_API_KEY, like every other substreams command:

export SUBSTREAMS_API_KEY="<GRAPHOPS_API_KEY>"

You pass the GraphOps Ethereum mainnet endpoint with -e when you run the sink.

Initialize tables

Run setup once per database. It creates the sink's system tables, cursors and substreams_history, and runs the package's schema.sql.

substreams sink postgres setup "substreams-template@v0.3.1" --dsn "$DSN"

The package schema creates application tables such as:

CREATE TABLE transfer (
  id TEXT NOT NULL,
  trx_hash TEXT NOT NULL,
  "from" TEXT NOT NULL,
  "to" TEXT NOT NULL,
  token_id BIGINT NOT NULL,
  ordinal BIGINT NOT NULL,
  PRIMARY KEY (id)
);

Run the sink

Run a bounded range first:

substreams sink postgres "substreams-template@v0.3.1" \
  -e "https://1.evm.substreams.graphops.xyz" \
  -s 12292922 \
  -t 12293022 \
  --dsn "$DSN"

This streams the package's configured SQL output module, writes rows to Postgres, and stores the sink cursor in the database. The package's db_out module is selected from the manifest's sink configuration:

sink:
  module: db_out
  type: sf.substreams.sink.sql.v1.Service
  config:
    schema: "./schema.sql"

To follow the chain after the initial range, omit the stop block:

substreams sink postgres "substreams-template@v0.3.1" \
  -e "https://1.evm.substreams.graphops.xyz" \
  -s 12292922 \
  --dsn "$DSN"

When you restart the sink against the same database, it resumes from the stored cursor.

Verify rows

Inspect the tables:

psql "$DSN" -c '\dt'

Count materialized transfer rows:

psql "$DSN" -c 'select count(*) as transfer_count from transfer;'

 transfer_countColumn alias from the query----------------             42Rows the sink committed for the range(1 row)

Illustrative psql output. Any count above zero shows the sink wrote rows; the exact number depends on the block range, package version, and chain data.

Read a few rows:

psql "$DSN" -c 'select trx_hash, "from", "to", token_id, ordinal from transfer order by ordinal limit 5;'

Use your own package

The sink chooses its mode from the output module's type. Your package must use one of them:

Mode	Requirement	Notes
Database Changes	The output module emits `sf.substreams.sink.database.v1.DatabaseChanges`, commonly from a `db_out` module, and the package provides `schema.sql`.	Supports inserts, updates and upserts.
Relational Mappings	The output Protobuf messages carry table and key annotations.	Insert-only; the schema is derived from the Protobuf. In beta: Postgres is stable, ClickHouse is in development.

For Database Changes, the package manifest names the module, engine and schema in its sink configuration:

sink:
  module: db_out
  type: sf.substreams.sink.sql.v1.Service
  config:
    engine: postgres
    schema: schema.sql

Then run:

substreams sink postgres setup "<PACKAGE_REF>" --dsn "$DSN"
substreams sink postgres "<PACKAGE_REF>" \
  -e "https://<CHAIN_ID>.evm.substreams.graphops.xyz" \
  -s <START_BLOCK> \
  -t <STOP_BLOCK> \
  --dsn "$DSN"

To sink a module other than the configured one, name it after the package reference. For ClickHouse, use substreams sink clickhouse with a clickhouse:// DSN; ClickHouse applies chain reorgs with a delay. Follow Substreams:SQL for both modes, Protobuf annotations and engine options.

If you ran the standalone substreams-sink-sql binary, the CLI keeps its cursor tables and schemas, so it resumes where the binary stopped. See Migrating from substreams-sink-sql for the command and flag changes.

Before production

Start with bounded ranges before running an open-ended live sink.
Use a dedicated database per sink.
Store the package reference, endpoint, range, and Substreams CLI version in release records.
Let the sink manage its cursor and history tables; back up those tables with the materialized data.
Keep package versions pinned in production.
Monitor blocks processed, bytes out, row flush failures, retries, and sink lag.