Розробка on-chain ETL-пайплайну

Розробка ETL-пайплайну для On-chain даних

Через три дні після запуску аналітики Uniswap v3 ви виявляєте, що eth_getLogs з широким фільтром почав таймаутити, агрегати розходяться через пропущені реорганізації, а ваш PostgreSQL припиняється на гігабайтних таблицях без розділення. On-chain ETL — це не просто "читаємо логи та пишемо в базу". Це система з гарантіями консистентності, обробкою реорганізацій, трансформацією даних та керованим буфером. Давайте побудуємо це правильно з самого початку.

Архітектура: Три шари ETL

Класичний ETL (Extract — Transform — Load) набуває специфіки в контексті блокчейну: джерело даних незмінне, але не остаточне (реорги), обсяги вимірюються сотнями мільйонів подій, а затримка може бути як секунди, так і години — залежить від завдання.

Extract: Надходження з вузла

Вибір джерела даних визначає все інше. Три рівні з зростаючою складністю:

• Logs/Events — те, що контракт явно видає. Дешево, швидко, структуровано через ABI. Обмеження: тільки те, що розробник вирішив логувати.
• Traces (внутрішні транзакції) — усі виклики всередині транзакції, включаючи ETH-передачі без подій. Потребує debug_traceTransaction або trace_block (Parity-style). Не всі вузли це підтримують; Erigon — найкращий вибір для завдань із трасуванням.
• State diffs — зміни слотів сховища за блоком. Максимальна повнота, але величезний обсяг даних і складність інтерпретації без ABI.

Для більшості DeFi-завдань достатньо logs + traces. State diffs потрібні для MEV-аналітики та моніторингу контрактів без подій (наприклад, legacy WETH).

Паттерни отримання даних:

# Polling з експоненціальним 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" — ділимо діапазон навпіл
        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

Цей паттерн рекурсивного бісекту обов'язковий. Публічні RPC (і навіть Alchemy/Infura) обрізають відповіді за розміром. Без цього ваш пайплайн впаде на активних блоках.

WebSocket підписки для реального часу: eth_subscribe("newHeads") дає нові блоки, eth_subscribe("logs", filter) потокує подій. Критично: завжди робіть catch-up через polling від останньої оброблено блоку при переконнекті.

Firehose (StreamingFast/Pinax) — бінарний протокол над gRPC, спеціально для high-throughput індексування. Швидкість надходження на порядок вища за JSON-RPC. Використовується в Substreams. Якщо вам потрібно обробити 15M+ блоків Ethereum — розглядайте це в першу чергу.

Transform: Трансформація та збагачення

Це найбільш логічно складний шар. Завдання:

Декодування ABI. Raw log — це topics[] (bytes32) та data (bytes). Декодування через viem/ethers/web3.py. Нюанс із proxy-контрактами: беріть ABI від implementation, а не від proxy. EIP-1967 визначає стандартний слот 0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc для адреси реалізації.

import { decodeEventLog, parseAbiItem } from 'viem'

// Для proxy: резолвимо 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 })

Збагачення даних. Raw події рідко містять все необхідне. Типові збагачення:

• USD-вартість: витягуємо ціну токена на момент блоку з Chainlink або власного price oracle
• Метадані токенів: symbol(), decimals() — кешуємо агресивно, вони незмінні
• Розпізнавання ідентичності: картування адрес на відомі протоколи (Uniswap Router, Aave Pool)

Нормалізація. Приводимо суми токенів до коректного масштабу decimal. uint256 із контракту → Python Decimal або PostgreSQL numeric — ніколи не float64, втратите точність на великих значеннях.

Трансформації зі станом — найскладніша частина. Обчислення поточних сум, поточних балансів, позицій LP. Потребує суворого порядку обробки подій всередині блоку (сортування за logIndex).

Load: Запис до сховища

Пакетний запис — обов'язково. Не INSERT по одному рядку. PostgreSQL COPY або bulk INSERT через executemany:

# У 10-50 разів швидше за одиничні INSERT
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 — страховка від дублів при retry після помилки. Завжди додавайте UNIQUE(tx_hash, log_index).

Обробка реорганізацій: Обов'язкова

Реорганізація на Ethereum — не винятковий випадок. На PoS Ethereum реорги глибиною 1-2 блоки трапляються кілька разів на день. Ігнорувати їх — означає мати "забруднені" дані у базі.

Стратегія: tombstone + replay. Кожен запис містить block_hash. При отриманні нового блоку перевіряємо, чи не змінився block_hash для вже оброблено block_number:

-- Виявлення реорга
SELECT block_number, block_hash
FROM processed_blocks
WHERE block_number >= $1
  AND block_hash != ANY($2::bytea[])
ORDER BY block_number;

При виявленні розбіжності — відкат та переобробка в одній транзакції:

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;

Для фінансових даних — чекаємо safe finality (12+ блоків на PoS Ethereum) перед тим як вважати дані достовірними. Для аналітики — latest достатньо з позначкою "попередній".

Черга та оркестрація

Для нетривіальних пайплайнів потрібна черга між Extract та Transform/Load — буфер при пиковому навантаженні та ізоляція відмов.

Для більшості DeFi-проектів Redis Streams достатньо. Kafka додає операційну складність, але дозволяє replay — переочитувати історію при додаванні нової трансформації.

Оркестрація з Airflow або Prefect потрібна, коли пайплайн має залежності: спочатку завантажуємо ціни, потім обчислюємо USD-вартість свопів. DAG явно описує ці залежності.

Схема бази даних

Критичні рішення щодо схеми:

Розділення за часом — обов'язково для таблиць подій. PostgreSQL native partitioning або TimescaleDB hypertables. Без розділення VACUUM на таблиці з 500M рядків займе години та заблокує inserts.

-- TimescaleDB: автоматичне розділення за часом
SELECT create_hypertable('swaps', 'block_time', chunk_time_interval => INTERVAL '1 day');
-- Стиснення старих chunks
SELECT add_compression_policy('swaps', INTERVAL '7 days');

Тільки потрібні індекси. Кожен індекс — це overhead на INSERT. Типовий набір:

• (pool_address, block_time) — запити для конкретного пулу за період
• (sender, block_time) — історія транзакцій користувача
• (tx_hash, log_index) — UNIQUE constraint для ідемпотентності

Матеріалізовані представлення для агрегатів. Не обчислюйте суми на 100M рядках на-льоту. Матеріалізоване представлення з щогодинними/щоденними агрегатами + REFRESH MATERIALIZED VIEW CONCURRENTLY за розписанням.

Продуктивність: Реальні числа

Для орієнтації: пайплайн на Python + asyncio + PostgreSQL на сервері 8 CPU / 32 GB RAM обробляє ~2000-5000 подій/сек при записі. Для історичної синхронізації Ethereum (2M+ блоків) це означає дні роботи.

Оптимізації в порядку впливу:

1. Паралельне надходження — кілька воркерів на різних діапазонах блоків. Лінійне прискорення до кількості CPU та RPC лімітів.
2. Вимкнення індексів під час bulk load — завантажуємо сирі дані, потім CREATE INDEX CONCURRENTLY. Прискорення вставки в 3-10 разів.
3. Перехід на Rust/Go для критичних компонентів. Парсинг ABI та десеріалізація блоків в Rust (alloy crate) у 10-20 разів швидше за Python.
4. Firehose замість JSON-RPC — якщо доступний для вашої мережі, дає прискорення надходження в 5-10 разів.

Моніторинг пайплайну

Метрики, які мають бути з першого дня:

• Pipeline lag — current_block - processed_block. Алерт при > 20 блоків. Зростання lag означає вузле місце.
• Reorg rate — кількість реорганізацій за годину. Різке зростання = нестабільна нода або RPC.
• Throughput — подій/сек на кожному етапі. Дозволяє знайти bottleneck.
• Error rate — кількість помилок декодування. > 0 означає невідомий ABI або змінений контракт.

# Grafana/Prometheus alert
- alert: ETLPipelineLagHigh
  expr: blockchain_latest_block - etl_processed_block > 50
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "ETL pipeline lag: {{ $value }} blocks"

Технологічний стек

Процес розробки

Фаза 1 (3-5 днів): Проектування. Визначаємо джерела даних, контракти та подій, схему БД, вимоги до затримки та обсягу. Прототип ingester на тестових даних.

Фаза 2 (7-14 днів): Ядро пайплайну. Extract + Transform + Load з обробкою реорганізацій. Тестування на mainnet-даних, верифікація коректності через порівняння зі станом блокчейну.

Фаза 3 (3-5 днів): Продуктивність. Профілювання, оптимізація bottleneck-ів, налаштування БД (індекси, розділення, vacuum).

Фаза 4 (2-3 дні): Deploy та моніторинг. Docker Compose або Kubernetes, налаштування алертів, runbook.

Усього: 2-4 тижні для пайплайну одного протоколу. Мультичейн з cross-chain агрегацією — 4-8 тижнів.