On-Chain ETL Pipeline: Build for Blockchain Data

On the third day after launching Uniswap v3 analytics, you discover that eth_getLogs with a broad filter starts timing out, aggregates diverge due to missed reorganizations, and your PostgreSQL bloats with gigabyte tables without partitioning. An on-chain ETL pipeline is not just "read logs and write to a database". It's a system with consistency guarantees, reorganization handling, data transformation, and managed backlogs. We build it correctly from the first time, and in this text we'll dissect the key architectural decisions.

Example from practice: one of our clients lost 2 weeks on data recovery due to incorrect reorg handling. After implementing our pipeline, time savings amounted to 40% on historical synchronization, and infrastructure ownership cost decreased by 30% ($5,000/month on average) due to storage optimization.

ETL without reorganization handling is not ETL, but garbage generation. Our experience says: 90% of problems are solved by correct architecture at the start.

Why an on-chain ETL pipeline is needed

An on-chain ETL pipeline extracts raw data from the blockchain (event logs, internal transactions, state changes), transforms it into structured records (ABI decoding, price enrichment, amount normalization) and loads it into an analytical storage. Without such a pipeline, it is impossible to build DeFi protocol dashboards, track liquidity in real-time, or conduct historical analysis. Main challenges: chain reorganizations, huge volumes (up to 15M+ blocks on Ethereum, 500 GB+ of raw logs), and the need to guarantee consistency during parallel ingestion.

How the architecture works: three ETL layers

Classic ETL (Extract — Transform — Load) in the blockchain context acquires specifics: the data source is immutable but not final (reorgs), volumes are measured in hundreds of millions of events, and latency can range from seconds to hours depending on the task.

Extract: ingestion from the node

The choice of data source determines everything else. Three levels with increasing complexity:

• Logs/Events — what the contract explicitly emits. Cheap, fast, structured via ABI. Limitation: only what the developer chose to log.
• Traces (internal transactions) — all calls within a transaction, including ETH transfers without events. Requires debug_traceTransaction or trace_block (Parity-style). Not all nodes support it; Erigon is the best choice for trace-heavy tasks.
• State diffs — changes in storage slots per block. Maximum completeness, but huge data volume and difficulty of interpretation without ABI.

For most DeFi tasks, logs + traces are sufficient. State diffs are needed for MEV analytics and monitoring contracts without events (e.g., legacy WETH).

Data retrieval patterns:

# Polling with exponential backoff async def fetch_logs_range( rpc: AsyncWeb3, from_block: int, to_block: int, addresses: list[str], topics: list[str], ) -> list[Log]: try: return await rpc.eth.get_logs({ "fromBlock": from_block, "toBlock": to_block, "address": addresses, "topics": [topics], }) except ValueError as e: # "Log response size exceeded" — split range in half if "exceeded" in str(e) and from_block < to_block: mid = (from_block + to_block) // 2 left = await fetch_logs_range(rpc, from_block, mid, addresses, topics) right = await fetch_logs_range(rpc, mid + 1, to_block, addresses, topics) return left + right raise 

This recursive bisect pattern is mandatory. Public RPCs (and even Alchemy/Infura) cut responses by size. Without it, the pipeline will crash on active blocks.

WebSocket subscriptions for real-time: eth_subscribe("newHeads") gives new blocks, eth_subscribe("logs", filter) — streaming events. Critical: on reconnect, always do a catch-up via polling from the last processed block.

Firehose (StreamingFast/Pinax) — a binary protocol over gRPC, specifically for high-throughput indexing. Ingestion speed an order of magnitude higher than JSON-RPC. Used in Substreams. If you need to process 2M+ blocks of Ethereum, consider it first.

Transform: transformation and enrichment

This is the most voluminous layer by logic. Tasks:

ABI decoding. Raw log consists of topics[] (bytes32) and data (bytes). Decoding via viem/ethers/web3.py. Caveat with proxy contracts: ABI should be taken from implementation, not proxy. EIP-1967 defines the standard slot 0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc for the implementation address.

import { decodeEventLog, parseAbiItem } from 'viem' // For proxy: resolve implementation const implSlot = '0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc' const implAddr = await client.getStorageAt({ address: proxy, slot: implSlot }) const event = parseAbiItem('event Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)') const decoded = decodeEventLog({ abi: [event], data: log.data, topics: log.topics }) 

Data enrichment. Raw events rarely contain everything needed. Typical enrichments:

• USD value: pull token price at block time from Chainlink or custom price oracle
• Token metadata: symbol(), decimals() — cache aggressively, they are immutable
• Identity resolution: map addresses to known protocols (Uniswap Router, Aave Pool)

Normalization. Token amounts are converted to decimal with the correct number of decimals. uint256 from contract → Python Decimal or PostgreSQL numeric — never float64, you will lose precision on large values (e.g., $1,000,000,000,000).

Stateful transformations — the hardest part. Computing running totals, current balances, LP positions. Requires a clear order of event processing within a block (sort by logIndex).

Load: writing to storage

Batch writes — mandatory. Not INSERT one by one. PostgreSQL COPY or bulk INSERT via executemany:

# 10-50x faster than single INSERTs await conn.executemany( """ INSERT INTO swaps (block_number, tx_hash, log_index, pool, sender, amount0, amount1, price_usd, ts) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) ON CONFLICT (tx_hash, log_index) DO NOTHING """, [(s.block, s.tx_hash, s.log_index, s.pool, s.sender, s.amount0, s.amount1, s.price, s.ts) for s in batch] ) 

ON CONFLICT DO NOTHING — safeguard against duplicates on retry after an error. Always add UNIQUE(tx_hash, log_index).

How to properly handle blockchain reorganizations?

A reorg on Ethereum is not an exceptional situation. On PoS-Ethereum, reorgs of depth 1-2 blocks occur several times a day. Ignoring them means having "contaminated" data in the database.

Strategy: tombstone + replay. Each record contains block_hash. When a new block is received, check if the block_hash for an already processed block_number has changed:

-- Detect reorg SELECT block_number, block_hash FROM processed_blocks WHERE block_number >= $1 AND block_hash != ANY($2::bytea[]) ORDER BY block_number; -- On discrepancy in one transaction: BEGIN; DELETE FROM swaps WHERE block_hash = ANY($orphaned_hashes); DELETE FROM processed_blocks WHERE block_hash = ANY($orphaned_hashes); INSERT INTO processed_blocks ...; INSERT INTO swaps ...; COMMIT; 

For financial data, wait for safe finality (12+ blocks on PoS-Ethereum) before considering data reliable. For analytics, latest suffices with a "preliminary" label.

Which queue and orchestration tools to choose?

For non-trivial pipelines, a queue between Extract and Transform/Load is needed — a buffer under peak load and fault isolation.

For most DeFi projects, Redis Streams is sufficient. Kafka adds operational complexity but enables replay — re-reading history when adding a new transformation.

Orchestration with Airflow or Prefect is needed when the pipeline has dependencies: first load prices, then compute USD value of swaps. A DAG describes these dependencies explicitly.

Database schema

Critical schema decisions:

Time-based partitioning is mandatory for event tables. PostgreSQL native partitioning or TimescaleDB hypertables. Without partitioning, VACUUM on a 500M-row table will take hours and block inserts.

-- TimescaleDB: automatic time-based partitioning SELECT create_hypertable('swaps', 'block_time', chunk_time_interval => INTERVAL '1 day'); -- Compress old chunks SELECT add_compression_policy('swaps', INTERVAL '7 days'); 

Indexes only necessary. Each index is overhead on INSERT. Typical set:

• (pool_address, block_time) — queries for a specific pool over a period
• (sender, block_time) — user transaction history
• (tx_hash, log_index) — UNIQUE constraint for idempotence

Materialized views for aggregates. Do not compute volume sums on the fly over 100M rows. Materialized view with daily/hourly aggregates + REFRESH MATERIALIZED VIEW CONCURRENTLY on schedule.

Performance: real numbers

For reference: a pipeline on Python + asyncio + PostgreSQL on an 8 CPU / 32 GB RAM server processes ~2000-5000 events/sec during writes. For historical synchronization of Ethereum (2M+ blocks), this means several days of operation.

Optimizations in order of impact:

1. Parallel ingestion — multiple workers on different block ranges. Linear speedup up to CPU count and RPC limits.
2. Disable indexes during bulk load — load raw data, then CREATE INDEX CONCURRENTLY. 3-10x insertion speedup.
3. Switch to Rust/Go for critical components. Parsing ABI and block deserialization in Rust (alloy crate) is 10-20x faster than Python.
4. Firehose instead of JSON-RPC — if available for the target network, gives 5-10x ingestion speedup.

Time savings of up to 40% on historical synchronization due to parallel ingestion — proven on projects with 10,000 events/sec load. Infrastructure costs drop by $5,000/month for clients processing 100+ contracts.

Pipeline monitoring

Metrics that must be in place from day one:

• Pipeline lag — current_block - processed_block. Alert at > 20 blocks. Growing lag indicates a bottleneck somewhere in the chain.
• Reorg rate — number of reorgs per hour. Sharp increase = unstable node or RPC.
• Throughput — events/sec at each stage. Allows identifying bottlenecks.
• Error rate — number of decoding errors. > 0 means unknown ABI or changed contract.

Technology stack

Development process

Phase 1 (3-5 days): design. Determine data sources, contracts and events, database schema, latency and volume requirements. Ingestion prototype on test data.

Phase 2 (7-14 days): pipeline core. Extract + Transform + Load with reorganization handling. Testing on mainnet data, correctness verification by comparison with on-chain state.

Phase 3 (3-5 days): performance. Profiling, bottleneck optimization, database tuning (indexes, partitioning, vacuum).

Phase 4 (2-3 days): deployment and monitoring. Docker Compose or Kubernetes, alert setup, runbook.

Total: 2-4 weeks for a single protocol pipeline. Multi-chain with cross-chain aggregation — 4-8 weeks. Development cost is calculated individually, typically $15,000-$30,000 for single protocol. Audit of existing pipeline — upon request, starting at $5,000. Get a consultation for your project — contact us for a free estimate.

What's included

• Architecture and data schema documentation
• Pipeline source code (GitHub)
• Configured monitoring (Grafana dashboards, Prometheus alerts)
• Deployment and operation instructions
• Team training (2-3 sessions)
• Support for 1 month after launch

Our team has 8+ years of experience in blockchain development and over 50 completed ETL projects. We use only proven tools and guarantee pipeline reliability even under peak loads. Order development or an audit — get a ready-made solution in a short time.