Pipeline cookbook

This page collects complete, runnable pipeline configurations for common use cases. Copy a recipe, change the name to something unique, swap in your own addresses and secret names, then validate and deploy:

goldsky turbo validate pipeline.yaml
goldsky turbo apply pipeline.yaml -i

Recipes use Ethereum Mainnet (and Base for multi-chain). Swap the dataset_name prefix for any other chain; see supported networks. For guidance on which shape fits your use case, see pipeline architecture patterns; for full field syntax, see the pipeline configuration reference.

Starters

Minimal ERC-20 stream

Use this to verify your setup end to end. It needs no credentials because the blackhole sink discards all data. Pair it with live inspection to watch data flow through.

name: my-erc20-pipeline
resource_size: s

sources:
  transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest # Change to 'earliest' to process historical data

transforms: {}

sinks:
  output:
    type: blackhole
    from: transfers

Filter transfers by token contract

Use this when you only care about specific tokens. The address column on erc20_transfers holds the token contract address; compare against a lowercased literal.

name: usdc-transfers
resource_size: s

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  usdc_only:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        sender,
        recipient,
        amount,
        address AS token_address,
        to_timestamp(block_timestamp) AS block_time,
        block_number
      FROM ethereum_transfers
      WHERE address = lower('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48')

sinks:
  output:
    type: blackhole
    from: usdc_only

Common Ethereum Mainnet addresses to swap in:

USDC: 0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48
WETH: 0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2
DAI: 0x6B175474E89094C44Da98b954EedeAC495271d0F

To track a set of tokens you can change without redeploying, use a dynamic table instead of a hardcoded address.

Solana token transfers

Use this to stream SPL token transfers. Solana sources use slot numbers (start_block) instead of start_at; see Solana sources.

Turbo only: Solana datasets are exclusively available with the Turbo (Streamling) engine. They are not available in Mirror v1 pipelines.

name: solana-transfers
resource_size: s

sources:
  sol_transfers:
    type: dataset
    dataset_name: solana.token_transfers
    version: 1.0.0
    # Solana positioning options:
    # - Omit start_block to start from the latest slot
    # - Add 'start_block: 250000000' to start from a specific slot

transforms: {}

sinks:
  output:
    type: blackhole
    from: sol_transfers

Shaping patterns

Linear: decode contract events into Postgres

Use this when one source flows through a chain of transforms into one sink. It is the worked example of the linear pattern. It decodes OrderFilled events from an exchange contract into typed rows.

name: exchange-trades
resource_size: s

sources:
  raw_logs:
    type: dataset
    dataset_name: ethereum.raw_logs
    version: 1.0.0
    start_at: earliest
    # Fast scan: pre-filter at the source so the backfill skips irrelevant blocks.
    # Replace the address with your contract.
    filter: >-
      address = '0x1111111111111111111111111111111111111111'
      AND block_number >= 21000000

transforms:
  # Step 1: decode raw logs with an inline ABI
  decoded_events:
    type: sql
    primary_key: id
    sql: |
      SELECT
        _gs_log_decode(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"maker","type":"address"},{"indexed":true,"name":"taker","type":"address"},{"indexed":false,"name":"amount","type":"uint256"},{"indexed":false,"name":"price","type":"uint256"}],"name":"OrderFilled","type":"event"}]',
          topics,
          data
        ) AS decoded,
        id, block_number, transaction_hash, address, block_timestamp
      FROM raw_logs

  # Step 2: extract typed columns from the decoded event
  trades:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        block_number,
        block_timestamp,
        transaction_hash,
        address AS contract_address,
        decoded.event_params[1] AS maker,
        decoded.event_params[2] AS taker,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        (CAST(decoded.event_params[4] AS DOUBLE) / 1e6) AS price_usdc
      FROM decoded_events
      WHERE decoded.event_signature = 'OrderFilled'

sinks:
  database:
    type: postgres
    from: trades
    schema: public
    table: trades
    secret_name: MY_POSTGRES_SECRET
    primary_key: id

Fan-out: one source, multiple destinations

Use this when different consumers need different views of the same data. It is the worked example of the fan-out pattern. USDC transfers go to a ClickHouse warehouse; high-value transfers of any token go to an alerting webhook.

name: transfer-fan-out
resource_size: m

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # View 1: USDC transfers, reshaped for analytics
  usdc_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        sender,
        recipient,
        (CAST(amount AS DOUBLE) / 1e6) AS amount_usdc,
        block_number,
        block_timestamp
      FROM ethereum_transfers
      WHERE address = lower('0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48')

  # View 2: high-value transfers of any token, for alerting
  whale_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        address AS token_address,
        sender,
        recipient,
        amount,
        transaction_hash,
        block_timestamp
      FROM ethereum_transfers
      WHERE amount > 1000000000000000000000000

sinks:
  # Sink 1: analytics warehouse
  warehouse:
    type: clickhouse
    from: usdc_transfers
    table: usdc_transfers
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 100000
    batch_flush_interval: 10s

  # Sink 2: real-time alerts
  alerts:
    type: webhook
    from: whale_transfers
    url: https://alerts.example.com/whale-transfer
    one_row_per_request: true

Fan-in: unified activity feed

Use this when several event types should land in one table with a shared schema. It is the worked example of the fan-in pattern. It decodes four WETH event types and combines them with UNION ALL.

WETH activity feed

name: weth-activity-feed
resource_size: l

sources:
  raw_logs:
    type: dataset
    dataset_name: ethereum.raw_logs
    version: 1.0.0
    start_at: earliest
    # Fast scan: only WETH logs, starting from a recent block
    filter: >-
      address = '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2'
      AND block_number >= 21000000

transforms:
  # Decode all four event types in a single transform
  decoded_events:
    type: sql
    primary_key: id
    sql: |
      SELECT
        _gs_log_decode(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"from","type":"address"},{"indexed":true,"name":"to","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Transfer","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"owner","type":"address"},{"indexed":true,"name":"spender","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Approval","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"dst","type":"address"},{"indexed":false,"name":"wad","type":"uint256"}],"name":"Deposit","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"src","type":"address"},{"indexed":false,"name":"wad","type":"uint256"}],"name":"Withdrawal","type":"event"}]',
          topics,
          data
        ) AS decoded,
        id, block_number, transaction_hash, address, block_timestamp
      FROM raw_logs

  # Event type 1: transfers
  transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        decoded.event_params[2] AS counterparty,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        'TRANSFER' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Transfer'

  # Event type 2: approvals
  approvals:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        decoded.event_params[2] AS counterparty,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e18) AS amount,
        'APPROVAL' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Approval'

  # Event type 3: deposits (ETH wrapped into WETH)
  deposits:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        '' AS counterparty,
        (CAST(decoded.event_params[2] AS DOUBLE) / 1e18) AS amount,
        'DEPOSIT' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Deposit'

  # Event type 4: withdrawals (WETH unwrapped back to ETH)
  withdrawals:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id, block_number, block_timestamp, transaction_hash, address,
        decoded.event_params[1] AS user_address,
        '' AS counterparty,
        (CAST(decoded.event_params[2] AS DOUBLE) / 1e18) AS amount,
        'WITHDRAWAL' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Withdrawal'

  # Combine all events into a unified activity feed.
  # Every branch must project the same columns in the same order.
  all_activity:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM transfers
      UNION ALL
      SELECT * FROM approvals
      UNION ALL
      SELECT * FROM deposits
      UNION ALL
      SELECT * FROM withdrawals

sinks:
  warehouse:
    type: clickhouse
    from: all_activity
    table: weth_activity
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 1000
    batch_flush_interval: 1s

Multi-chain: combine Ethereum and Base

Use this for cross-chain analytics where multiple chains land in a single stream. Add a source per chain and another UNION ALL branch; use m for two chains and l as you add more.

name: multi-chain-transfers
resource_size: m # Multiple sources need more headroom

sources:
  ethereum_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

  base_transfers:
    type: dataset
    dataset_name: base.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Tag each row with its chain before combining
  combined:
    type: sql
    primary_key: id
    sql: |
      SELECT *, 'ethereum' AS chain FROM ethereum_transfers
      UNION ALL
      SELECT *, 'base' AS chain FROM base_transfers

sinks:
  output:
    type: blackhole
    from: combined

Replace the blackhole sink with a real destination for production. If the chains don’t need to share one table, prefer the templated per-chain recipe below.

Multi-chain templated: one pipeline per chain

Use this when you run the same logic on several chains but want independent deployment, checkpointing, and monitoring per chain. It is the worked example of the templated deployment pattern, which also lists exactly which values to swap per chain.

ethereum-transfers.yaml

name: ethereum-transfer-streaming
resource_size: m

sources:
  ethereum_erc20_transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Tag every row with its chain so downstream tables can be unioned later
  chain_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        'ethereum' AS chain,
        address AS token_address,
        sender,
        recipient,
        amount,
        block_number,
        block_timestamp
      FROM ethereum_erc20_transfers

sinks:
  warehouse:
    type: clickhouse
    from: chain_transfers
    table: ethereum_token_transfers
    secret_name: MY_CLICKHOUSE
    primary_key: id
    batch_size: 100000
    batch_flush_interval: 10s

To deploy the Base copy, swap ethereum for base in the pipeline name, source key, dataset_name, the SQL chain literal, and the sink table, then apply both files.

Sinks and delivery

Stream transfers to PostgreSQL

Use this to land a dataset in your application database with no transformation. The table is created automatically if it doesn’t exist. Create the secret with goldsky secret create MY_POSTGRES_SECRET; see the PostgreSQL sink page for hosted and bring-your-own database options.

name: transfers-to-postgres
resource_size: s

sources:
  transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms: {}

sinks:
  postgres_output:
    type: postgres
    from: transfers
    schema: public
    table: erc20_transfers
    secret_name: MY_POSTGRES_SECRET # Change this to your secret name
    primary_key: id

For running totals instead of raw rows, see the PostgreSQL aggregate sink pattern.

Publish to Google Cloud Pub/Sub

Use this to feed GCP-based consumers. Before deploying (full details in the Pub/Sub sink reference):

Create the topic in your GCP project; Goldsky does not auto-create topics.
Grant the service account both roles/pubsub.publisher and roles/pubsub.viewer. The sink verifies the topic exists at startup, so a publish-only service account fails to initialize.
Store the project ID and service-account JSON as a secret: goldsky secret create --name MY_PUBSUB_SECRET --type pubsub.

name: erc20-transfers-pubsub
resource_size: s

sources:
  transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms: {}

sinks:
  pubsub_output:
    type: pubsub
    from: transfers
    topic: erc20-transfers # Must already exist in your GCP project
    secret_name: MY_PUBSUB_SECRET
    # Optional batching:
    # batch_size: 1000
    # batch_flush_interval: 1s

Archive, alert, and stream from one pipeline

Use this when the same source feeds several destinations at once: archive everything to Postgres, alert on large transfers via webhook, and stream to Kafka for event-driven consumers. Remove any sink you don’t need. Each sink needs its own secret; see the PostgreSQL, webhook, and Kafka sink pages.

name: transfers-multi-sink
resource_size: m

sources:
  transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Filter for large transfers (more than 1 whole token at 18 decimals)
  large_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM transfers
      WHERE amount > 1000000000000000000

sinks:
  # Sink 1: archive every transfer to PostgreSQL
  postgres_archive:
    type: postgres
    from: transfers
    schema: public
    table: all_transfers
    secret_name: MY_POSTGRES
    primary_key: id

  # Sink 2: send large transfers to a webhook for alerting
  webhook_alerts:
    type: webhook
    from: large_transfers
    url: https://api.example.com/alerts
    secret_name: MY_WEBHOOK # Optional; remove for unauthenticated endpoints

  # Sink 3: stream every transfer to Kafka (remove if not needed)
  kafka_stream:
    type: kafka
    from: transfers
    topic: erc20.transfers
    data_format: json
    secret_name: MY_KAFKA

Note that sinks can read directly from a source (from: transfers) or from a transform (from: large_transfers). For when to use one pipeline with multiple sinks versus separate pipelines, see the fan-out pattern.

Transforms

Multi-event activity feed

Use this as a compact starting point for decoding multiple events from one contract into a unified table, here USDC Transfer and Approval events. It’s a smaller version of the fan-in pattern; adapt the ABI, address, and column mappings for your contract.

name: usdc-activity-feed
resource_size: s

sources:
  raw_logs:
    type: dataset
    dataset_name: ethereum.raw_logs
    version: 1.0.0
    start_at: latest
    # Pre-filter at the source for efficiency (USDC on Ethereum)
    filter: >-
      address = '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48'

transforms:
  # Step 1: decode raw logs using the contract ABI
  decoded_events:
    type: sql
    primary_key: id
    sql: |
      SELECT
        _gs_log_decode(
          '[{"anonymous":false,"inputs":[{"indexed":true,"name":"from","type":"address"},{"indexed":true,"name":"to","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Transfer","type":"event"},{"anonymous":false,"inputs":[{"indexed":true,"name":"owner","type":"address"},{"indexed":true,"name":"spender","type":"address"},{"indexed":false,"name":"value","type":"uint256"}],"name":"Approval","type":"event"}]',
          topics,
          data
        ) AS decoded,
        id,
        block_number,
        transaction_hash,
        address,
        block_timestamp
      FROM raw_logs

  # Step 2a: extract Transfer events (USDC has 6 decimals)
  transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        block_number,
        block_timestamp,
        transaction_hash,
        address AS contract_address,
        decoded.event_params[1] AS from_address,
        decoded.event_params[2] AS to_address,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e6) AS amount,
        'TRANSFER' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Transfer'

  # Step 2b: extract Approval events (same output schema as transfers)
  approvals:
    type: sql
    primary_key: id
    sql: |
      SELECT
        id,
        block_number,
        block_timestamp,
        transaction_hash,
        address AS contract_address,
        decoded.event_params[1] AS from_address,
        decoded.event_params[2] AS to_address,
        (CAST(decoded.event_params[3] AS DOUBLE) / 1e6) AS amount,
        'APPROVAL' AS event_type
      FROM decoded_events
      WHERE decoded.event_signature = 'Approval'

  # Step 3: combine into a single stream
  all_activity:
    type: sql
    primary_key: id
    sql: |
      SELECT * FROM transfers
      UNION ALL
      SELECT * FROM approvals

sinks:
  output:
    type: blackhole
    from: all_activity
    # Replace with a real sink for production, for example:
    # type: postgres
    # schema: public
    # table: activities
    # secret_name: MY_POSTGRES_SECRET
    # primary_key: id

See the SQL functions reference for _gs_log_decode and other decoding helpers.

Throttle a high-volume webhook

Use this when a downstream service can’t keep up with the raw stream or enforces a rate limit. The throttle transform caps throughput at roughly max_batch_size / min_batch_interval, here about 10 records per second. Place the throttle just before the rate-limited sink so upstream transforms still run at full speed. See throttle transforms for details.

name: throttled-transfer-webhook
resource_size: s

sources:
  transfers:
    type: dataset
    dataset_name: ethereum.erc20_transfers
    version: 1.2.0
    start_at: latest

transforms:
  # Cap throughput at ~10 records/second (100 records / 10s)
  throttled:
    type: throttle
    from: transfers
    max_batch_size: 100
    min_batch_interval: 10s

sinks:
  webhook_output:
    type: webhook
    from: throttled
    url: https://api.example.com/transfers

Turbo

Guides

Reference

Mirror (legacy)

Starters

Minimal ERC-20 stream

Filter transfers by token contract

Solana token transfers

Shaping patterns

Linear: decode contract events into Postgres

Fan-out: one source, multiple destinations

Fan-in: unified activity feed

Multi-chain: combine Ethereum and Base

Multi-chain templated: one pipeline per chain

Sinks and delivery

Stream transfers to PostgreSQL

Publish to Google Cloud Pub/Sub

Archive, alert, and stream from one pipeline

Transforms

Multi-event activity feed

Throttle a high-volume webhook

​Starters

​Minimal ERC-20 stream

​Filter transfers by token contract

​Solana token transfers

​Shaping patterns

​Linear: decode contract events into Postgres

​Fan-out: one source, multiple destinations

​Fan-in: unified activity feed

​Multi-chain: combine Ethereum and Base

​Multi-chain templated: one pipeline per chain

​Sinks and delivery

​Stream transfers to PostgreSQL

​Publish to Google Cloud Pub/Sub

​Archive, alert, and stream from one pipeline

​Transforms

​Multi-event activity feed

​Throttle a high-volume webhook

Starters

Minimal ERC-20 stream

Filter transfers by token contract

Solana token transfers

Shaping patterns

Linear: decode contract events into Postgres

Fan-out: one source, multiple destinations

Fan-in: unified activity feed

Multi-chain: combine Ethereum and Base

Multi-chain templated: one pipeline per chain

Sinks and delivery

Stream transfers to PostgreSQL

Publish to Google Cloud Pub/Sub

Archive, alert, and stream from one pipeline

Transforms

Multi-event activity feed

Throttle a high-volume webhook