# Run a sink

Source: https://docs.graphops.xyz/substreams/guides/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._

- Package: `substreams-template@v0.3.1` (Output module db_out)
- GraphOps endpoint: `1.evm.substreams.graphops.xyz`
- Postgres: your database, tables `transfer` and `cursors`

1. setup: `sink postgres setup`, tables and cursor storage
2. run: `-s 12292922 -t 12293022`, rows and cursor per block
3. verify: `select count(*)`, rows in transfer

## Prerequisites

* An API key that includes Substreams.
* 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](https://docs.substreams.dev/how-to-guides/installing-the-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.

Tested with postgres 16-alpine\@sha256:721873c34ceb9f8d8fc265984940dc982404c105f19ad51be9fdc5970a6080ea.

```sh
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`:

```sh
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](https://docs.substreams.dev/reference-material/sql/sql/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:

```sh
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`.

Prerequisites

1. Install the Substreams CLI.

Tested with substreams-template v0.3.1, substreams 1.25.0.

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

```

The package schema creates application tables such as:

```sql
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:

Prerequisites

1. Install the Substreams CLI.

Tested with substreams-template v0.3.1, substreams 1.25.0.

```sh
substreams sink postgres "substreams-template@v0.3.1" \
  -e "https://1.evm.substreams.graphops.xyz" \
  --start-block 12292922 \
  --stop-block 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:

```yaml
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:

Prerequisites

1. Install the Substreams CLI.

Tested with substreams-template v0.3.1, substreams 1.25.0.

```sh
substreams sink postgres "substreams-template@v0.3.1" \
  -e "https://1.evm.substreams.graphops.xyz" \
  --start-block 12292922 \
  ${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:

```sh
psql "$DSN" -c '\dt'

```

Count materialized transfer rows:

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

```

Output

```text
 transfer_count 
----------------
              1
(1 row)

```

Column alias from the registered query

Rows committed by the local sink for the range

Read a few rows:

```sh
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:

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

```

Then run:

Prerequisites

1. Replace `<PACKAGE_REF>` with your Substreams package URL, `<START_BLOCK>` with the first block to process, and `<STOP_BLOCK>` with the block to stop before.

```sh
substreams sink postgres setup "<PACKAGE_REF>" --dsn "$DSN"
substreams sink postgres "<PACKAGE_REF>" \
  -e "$GRAPHOPS_ENDPOINT" \
  --start-block <START_BLOCK> \
  --stop-block <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](https://docs.substreams.dev/how-to-guides/sinks/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](https://docs.substreams.dev/how-to-guides/sinks/sql/migration)
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

* [Sinks and materialization](/substreams/concepts/sinks-and-materialization): Choose a sink for files, Pub/Sub, or ClickHouse.
* [Cursors and reorgs](/substreams/concepts/cursors-and-reorgs): Understand how the sink's cursor and history tables keep rows correct.
* [Substreams:SQL](https://docs.substreams.dev/how-to-guides/sinks/sql): Look up SQL sink flags, Protobuf annotations, and engine options.