Pipeline architecture patterns

Every Turbo pipeline has the same three building blocks (sources, transforms, and sinks), but how you arrange them determines how easy the pipeline is to operate, scale, and debug. This guide shows six patterns that cover most real-world pipelines, each with a complete, runnable example. Examples use Ethereum Mainnet (and Base for multi-chain). The same patterns work on any chain; see supported networks.

Choosing a pattern

Pattern	Use it when
Linear	One source, one destination, straightforward processing
Fan-out	One source feeding multiple destinations with different views
Fan-in	Multiple event types or chains combined into one table
Multi-chain templated	The same logic deployed per chain, with independent lifecycles
Dynamic tables	Filtering or enrichment against lookup data that changes at runtime
PostgreSQL aggregate sink	Real-time running totals (balances, counters) maintained in Postgres

Two decisions apply to every pattern:

Streaming or job mode: all examples below are streaming pipelines. For one-time backfills or exports, add job: true; see Job mode.
Resource size: each section suggests a starting resource_size. Start small and scale up if the pipeline lags; see the resource size reference for CPU and memory per tier.

Validate any pipeline before deploying it:

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

Linear pipeline

The simplest shape: one source, one or more chained transforms, one sink.

source → transform_a → transform_b → sink

Use when you have a single data source, a single destination, and straightforward processing: decode, filter, reshape. This example decodes OrderFilled events from an exchange contract and writes clean, typed rows to PostgreSQL:

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

Resource size: s for filtered streams; move to m if you backfill a busy contract from earliest. Related pages:

SQL transforms for the transform syntax used here
EVM sources for how filter speeds up backfills
PostgreSQL sink for sink options and secret format

Fan-out (one source, multiple sinks)

One source feeds multiple transforms, each writing a different view of the data to a different sink.

              ┌─→ transform_a ─→ sink_1 (ClickHouse)
source ───────┤
              └─→ transform_b ─→ sink_2 (webhook)

Use when different consumers need different views or subsets of the same data: for example, full history in a warehouse plus real-time alerts.

Default to one pipeline with multiple sinks when you send the same source data to several destinations. Sinks run independently: one failing does not block the others, and each can have its own batching settings. Splitting into one pipeline per destination duplicates source ingestion and wastes resources. Split only when the destinations need different resource sizes or genuinely independent lifecycles.

This example reshapes USDC transfers for an analytics warehouse and sends high-value transfers of any token 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

Resource size: m; multiple sinks mean more concurrent work than a linear pipeline. Related pages:

Multiple sinks for how sinks behave independently
ClickHouse sink and webhook sink for sink-specific options

Fan-in (multiple inputs, one sink)

Multiple event types are decoded from the same source, normalized to a common schema, and combined with UNION ALL into a single sink.

                  ┌─→ event_type_a ──┐
source → decode ──┤                  ├─→ UNION ALL → sink
                  └─→ event_type_b ──┘

Use when you want a unified table combining several event types (trades, deposits, withdrawals, transfers) as one activity feed. This example builds a WETH activity feed from four event types. Every branch projects the same columns in the same order, which UNION ALL requires:

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
  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

Resource size: l; decoding plus many transforms is the heaviest shape in this guide.

Multi-chain fan-in

The same shape works with multiple sources instead of multiple event types: combine chains into one output with UNION ALL.

ethereum_transfers ──┐
                     ├─→ UNION ALL → sink
base_transfers ──────┘

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:
  combined:
    type: sql
    primary_key: id
    sql: |
      SELECT *, 'ethereum' AS chain FROM ethereum_transfers
      UNION ALL
      SELECT *, 'base' AS chain FROM base_transfers

See the multi-chain recipe in the cookbook for the complete pipeline. Use m for two chains and l as you add more. If the chains do not need to land in the same table, prefer the templated deployment below.

Multi-chain templated deployment

When you need the same pipeline logic across multiple chains, deploy one pipeline file per chain rather than a single multi-source pipeline.

ethereum-transfers.yaml ─→ [pipeline 1] ─→ ethereum_token_transfers
base-transfers.yaml     ─→ [pipeline 2] ─→ base_token_transfers

Use when you want per-chain independence:

Independent lifecycle: deploy, pause, or delete one chain without touching the others
Independent checkpointing: one chain failing or lagging doesn’t block the others
Clearer monitoring: each chain has its own pipeline status and logs

The pattern is a template: write the pipeline once, then copy it and swap the chain-specific values. Here’s the Ethereum version:

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 create the Base version, copy the file and swap five values:

Field	Ethereum pipeline	Base pipeline
`name`	`ethereum-transfer-streaming`	`base-transfer-streaming`
Source key	`ethereum_erc20_transfers`	`base_erc20_transfers`
`dataset_name`	`ethereum.erc20_transfers`	`base.erc20_transfers`
Transform SQL	`'ethereum' AS chain`	`'base' AS chain`
Sink `table`	`ethereum_token_transfers`	`base_token_transfers`

Templated vs. multi-source

Approach	Pros	Cons
Templated (per-chain)	Independent lifecycle, clear monitoring	More files to manage
Multi-source (one file)	Single deployment, cross-chain UNION possible	Coupled lifecycle, harder to debug

Resource size: m per chain. Because each chain runs in its own pipeline, you can also size each one independently: a busy chain can run l while a quiet one runs s.

Dynamic table architecture

Dynamic tables give a pipeline runtime-updatable lookup data, the Turbo answer to “no joins in streaming SQL”. A dynamic_table transform is backed by a table (typically PostgreSQL) that you can update at any time without restarting the pipeline, and SQL transforms query it with the dynamic_table_check() function.

Pattern: dynamic allowlist or blocklist

                    ┌──────────────────────┐
                    │  External updates    │
                    │  (Postgres / API)    │
                    └──────────┬───────────┘
                               ▼
source ──→ sql transform ──→ [dynamic_table_check()] ──→ sink

The SQL transform filters records against the dynamic table. Change the table contents externally (from a backend, an admin tool, or plain SQL) and the pipeline picks up the change within seconds. Use when the set of things you filter on (wallets, contracts, tokens) changes over time, or is too large for a hardcoded WHERE ... IN list. This example only keeps transfers that touch a tracked wallet:

name: tracked-wallet-transfers
resource_size: s

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

transforms:
  # Postgres-backed lookup table; update it externally at any time
  tracked_wallets:
    type: dynamic_table
    backend_type: Postgres
    backend_entity_name: user_wallets
    secret_name: MY_POSTGRES

  # Keep only transfers involving a tracked wallet, and tag the direction
  wallet_transfers:
    type: sql
    primary_key: id
    sql: |
      SELECT
        *,
        CASE
          WHEN dynamic_table_check('tracked_wallets', sender) THEN 'outgoing'
          ELSE 'incoming'
        END AS direction
      FROM ethereum_transfers
      WHERE dynamic_table_check('tracked_wallets', sender)
         OR dynamic_table_check('tracked_wallets', recipient)

sinks:
  postgres_output:
    type: postgres
    from: wallet_transfers
    schema: public
    table: wallet_activity
    secret_name: MY_POSTGRES
    primary_key: id

Add a wallet from any Postgres client, and the pipeline starts matching it within seconds:

INSERT INTO streamling.user_wallets (value) VALUES (lower('0x...your-wallet...'));

Pattern: lookup enrichment

source ──→ decode ──→ filter ──→ sql (with dynamic_table_check) ──→ sink
                                        ▲
                              [token_metadata table]
                              (Postgres-backed)

Store lookup data (token allowlists, protocol contract sets, factory-created pool addresses) in the dynamic table and reference it from transforms. A dynamic table can even populate itself from pipeline data using an inline sql: query; see the factory pattern example.

Choosing a backend

Backend	`backend_type`	When to use
PostgreSQL	`Postgres`	Data managed by external systems; must persist across pipeline restarts
In-memory	`InMemory`	Auto-populated from pipeline data; ephemeral; avoids a database round trip

Use Postgres for production: it persists across restarts and can be updated externally.

Sizing considerations

Dynamic tables add memory overhead proportional to table size, so for large lookup tables (over ~100K rows) use the Postgres backend.
Lookups are batched and indexed, but cost still scales with table size, so keep tables as small as your use case allows.
Resource size: s is usually enough for allowlist filtering; size for your transform complexity, not for the dynamic table itself.

For full configuration syntax, table management, and more examples, see Dynamic tables and the dynamic_table_check reference.

PostgreSQL aggregate sink pattern

The postgres_aggregate sink maintains real-time running aggregations (balances, counters, totals) using a two-table pattern: a landing table receives raw events, and a database trigger incrementally updates an aggregation table.

source ──→ transform ──→ landing table ──→ trigger ──→ aggregation table

Use when you need running totals updated on every event, without recomputing aggregates from scratch. Your application reads the small aggregation table instead of scanning event history. This example turns every ERC-20 transfer into two balance changes (negative for the sender, positive for the recipient) and lets the trigger maintain per-account, per-token balances:

name: token-balances
resource_size: m

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

transforms:
  # Each transfer produces two rows: -amount for the sender, +amount for the recipient
  balance_changes:
    type: sql
    primary_key: change_id
    sql: |
      SELECT
        CONCAT(id, '-out') AS change_id,
        lower(sender) AS account,
        lower(address) AS token_address,
        CAST(amount AS DECIMAL(38, 0)) * -1 AS amount
      FROM ethereum_transfers
      UNION ALL
      SELECT
        CONCAT(id, '-in') AS change_id,
        lower(recipient) AS account,
        lower(address) AS token_address,
        CAST(amount AS DECIMAL(38, 0)) AS amount
      FROM ethereum_transfers

sinks:
  balances:
    type: postgres_aggregate
    from: balance_changes
    schema: public
    landing_table: balance_change_log
    agg_table: account_balances
    primary_key: change_id
    secret_name: MY_POSTGRES
    group_by:
      account:
        type: text
      token_address:
        type: text
    aggregate:
      balance:
        from: amount
        fn: sum
        type: numeric(38,0)
      change_count:
        fn: count

Query the result like any Postgres table:

SELECT balance FROM public.account_balances
WHERE account = lower('0x...') AND token_address = lower('0x...');

Supported aggregation functions: sum, count, avg, min, max. Not every function supports every operation type: sum and avg cannot handle updates, and min/max are insert-only. See supported aggregation functions before choosing.

This example tracks every token and holder on Ethereum Mainnet, which is a high write volume for a Postgres database. For production, add a WHERE clause to track specific tokens, or size your database accordingly.

Resource size: m. Throughput is usually bounded by the Postgres database rather than the pipeline, so scale the database before scaling the pipeline. For landing-table deduplication, trigger internals, and update/delete semantics, see the PostgreSQL aggregation sink reference.

Next steps

Copy-paste starting points for all of these patterns (and more) live in the pipeline cookbook.
Full YAML field reference: pipeline configuration.
Deploy and monitor with the CLI reference.

Turbo

Guides

Reference

Mirror (legacy)

Pipeline architecture patterns

Choosing a pattern

Linear pipeline

Fan-out (one source, multiple sinks)

Fan-in (multiple inputs, one sink)

Multi-chain fan-in

Multi-chain templated deployment

Templated vs. multi-source

Dynamic table architecture

Pattern: dynamic allowlist or blocklist

Pattern: lookup enrichment

Choosing a backend

Sizing considerations

PostgreSQL aggregate sink pattern

Next steps

​Choosing a pattern

​Linear pipeline

​Fan-out (one source, multiple sinks)

​Fan-in (multiple inputs, one sink)

​Multi-chain fan-in

​Multi-chain templated deployment

​Templated vs. multi-source

​Dynamic table architecture

​Pattern: dynamic allowlist or blocklist

​Pattern: lookup enrichment

​Choosing a backend

​Sizing considerations

​PostgreSQL aggregate sink pattern

​Next steps

Choosing a pattern

Linear pipeline

Fan-out (one source, multiple sinks)

Fan-in (multiple inputs, one sink)

Multi-chain fan-in

Multi-chain templated deployment

Templated vs. multi-source

Dynamic table architecture

Pattern: dynamic allowlist or blocklist

Pattern: lookup enrichment

Choosing a backend

Sizing considerations

PostgreSQL aggregate sink pattern

Next steps