> ## Documentation Index
> Fetch the complete documentation index at: https://docs.goldsky.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Pipeline cookbook

> Copy-paste-ready Turbo pipeline configurations for common use cases

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:

```bash theme={"dark"}
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](/chains/supported-networks#turbo). For guidance on which shape fits your use case, see [pipeline architecture patterns](/turbo-pipelines/guides/architecture-patterns); for full field syntax, see the [pipeline configuration reference](/turbo-pipelines/pipeline-config).

## Starters

### Minimal ERC-20 stream

Use this to verify your setup end to end. It needs no credentials because the [blackhole sink](/turbo-pipelines/sinks/blackhole) discards all data. Pair it with [live inspection](/turbo-pipelines/live-inspect) to watch data flow through.

```yaml theme={"dark"}
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.

```yaml theme={"dark"}
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](/turbo-pipelines/transforms/dynamic-tables) 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-pipelines/sources/solana).

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

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#linear-pipeline). It decodes `OrderFilled` events from an exchange contract into typed rows.

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#fan-out-one-source-multiple-sinks). USDC transfers go to a ClickHouse warehouse; high-value transfers of any token go to an alerting webhook.

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#fan-in-multiple-inputs-one-sink). It decodes four WETH event types and combines them with `UNION ALL`.

```yaml WETH activity feed expandable theme={"dark"}
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.

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#multi-chain-templated-deployment), which also lists exactly which values to swap per chain.

```yaml ethereum-transfers.yaml theme={"dark"}
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](/turbo-pipelines/sinks/postgres) page for hosted and bring-your-own database options.

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#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](/turbo-pipelines/sinks/pubsub) reference):

1. Create the topic in your GCP project; Goldsky does not auto-create topics.
2. 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.
3. Store the project ID and service-account JSON as a secret: `goldsky secret create --name MY_PUBSUB_SECRET --type pubsub`.

```yaml theme={"dark"}
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](/turbo-pipelines/sinks/postgres), [webhook](/turbo-pipelines/sinks/webhook), and [Kafka](/turbo-pipelines/sinks/kafka) sink pages.

```yaml theme={"dark"}
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](/turbo-pipelines/guides/architecture-patterns#fan-out-one-source-multiple-sinks).

## 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](/turbo-pipelines/guides/architecture-patterns#fan-in-multiple-inputs-one-sink); adapt the ABI, address, and column mappings for your contract.

```yaml theme={"dark"}
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](/turbo-pipelines/reference/sql-functions) 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](/turbo-pipelines/transforms/throttle) for details.

```yaml theme={"dark"}
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
```


## Related topics

- [Pipeline architecture patterns](/turbo-pipelines/guides/architecture-patterns.md)
- [Stream onchain data with Turbo](/turbo-pipelines/introduction.md)
- [Operating pipelines](/turbo-pipelines/guides/operating-pipelines.md)
- [Turbo Pipelines (turbo)](/compose/context/turbo.md)
- [Deploy a Turbo pipeline](/turbo-pipelines/quickstart.md)