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.

Packagesubstreams-template@v0.3.1Output module db_out

GraphOps1.evm.substreams.graphops.xyz

PostgresYour databasetransfer · cursors

You run

1. 01 · setupsink postgres setupTables and cursor storage
2. 02 · run-s 12292922 -t 12293022Rows and cursor per block
3. 03 · verifyselect count(*)Rows in transfer

Prerequisites

• A Substreams API key.
• The Substreams CLI. Install it with Download substreams 1.25.0 from https://github.com/streamingfast/substreams/releases/tag/v1.25.0 and verify the release checksum., or follow Install Substreams CLI(opens in a new tab).
• 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 -d --name "${POSTGRES_CONTAINER:-graphops-substreams-postgres}" \
  -e POSTGRES_USER=dev-node \
  -e POSTGRES_PASSWORD=local-dev-only-password \
  -e POSTGRES_DB=dev-node \
  -p "${POSTGRES_BINDING:-5432:5432}" \
  --health-cmd "pg_isready -h 127.0.0.1 -U dev-node -d dev-node" \
  public.ecr.aws/docker/library/postgres:16-alpine@sha256:721873c34ceb9f8d8fc265984940dc982404c105f19ad51be9fdc5970a6080ea

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

export DSN="postgres://dev-node:local-dev-only-password@localhost:${POSTGRES_PORT:-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(opens in a new tab) 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" \
  --start-block 12292922 \
  --stop-block 12293022 \
  --max-retries 0 \
  --force-protocol-version 2 \
  --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" \
  --start-block 12292922 \
  --max-retries 0 \
  --force-protocol-version 2 \
  ${STOP_BLOCK:+--stop-block "$STOP_BLOCK"} \
  --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_count Column alias from the registered query----------------              1Rows committed by the local sink for the range(1 row) 

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:

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 "$GRAPHOPS_ENDPOINT" \
  --start-block <START_BLOCK> \
  --stop-block <STOP_BLOCK> \
  --max-retries 0 \
  --force-protocol-version 2 \
  --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(opens in a new tab) 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(opens in a new tab) 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.

Set STOP_BLOCK to an exclusive block height when you want the continuous command to finish after a bounded range. Leave it unset to follow the chain.

For a different package, set GRAPHOPS_ENDPOINT to its chain endpoint and use a package that emits DatabaseChanges.

Next steps