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

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

На третий день после запуска аналитики по Uniswap v3 обнаруживаешь, что eth_getLogs с широким фильтром начинает таймаутить, агрегаты расходятся из-за пропущенных реорганизаций, а твой PostgreSQL пухнет с гигабайтными таблицами без партиционирования. On-chain ETL — это не просто "читаем логи и пишем в базу". Это система с гарантиями консистентности, обработкой реорганизаций, трансформацией данных и управляемым бэклогом. Строим её правильно с первого раза.

Архитектура: три слоя ETL

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

Extract: ингестия из ноды

Выбор источника данных определяет всё остальное. Три уровня с нарастающей сложностью:

• Logs/Events — то, что контракт явно эмитирует. Дёшево, быстро, структурировано через ABI. Ограничение: только то, что разработчик решил логировать.
• Traces (internal transactions) — все вызовы внутри транзакции, включая ETH-переводы без событий. Требует debug_traceTransaction или trace_block (Parity-style). Не все ноды поддерживают; Erigon — лучший выбор для trace-heavy задач.
• State diffs — изменения storage slot-ов по блоку. Максимальная полнота, но огромный объём данных и сложность интерпретации без 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 subscriptions для реалтайма: 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 определяет стандартный slot 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 })

Обогащение данных (enrichment). Чистые события редко содержат всё нужное. Типичные обогащения:

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

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

Трансформации с состоянием — самое сложное. Вычисление running total, текущих балансов, позиций LP. Требует чёткого порядка обработки событий внутри блока (сортировка по logIndex).

Load: запись в хранилище

Батчевая запись — обязательно. Не INSERT по одной записи. PostgreSQL COPY или bulk INSERT через executemany:

# 10-50x быстрее одиночных 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 достаточно с пометкой "preliminary".

Очередь и оркестрация

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

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

Оркестрация с Airflow или Prefect нужна когда пайплайн имеет зависимости: сначала загрузи цены, потом считай USD-стоимость свапов. DAG описывает эти зависимости явно.

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

Критические решения по схеме:

Партиционирование по времени — обязательно для таблиц событий. PostgreSQL native partitioning или TimescaleDB hypertables. Без партиционирования VACUUM на таблице из 500M строк займёт часы и заблокирует инсерты.

-- TimescaleDB: автоматическое партиционирование по времени
SELECT create_hypertable('swaps', 'block_time', chunk_time_interval => INTERVAL '1 day');
-- Компрессия старых чанков
SELECT add_compression_policy('swaps', INTERVAL '7 days');

Индексы только нужные. Каждый индекс — это overhead на INSERT. Типичный набор:

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

Materialized views для агрегатов. Не считайте суммы объёмов на лету по 100M строк. Materialized view с daily/hourly агрегатами + REFRESH MATERIALIZED VIEW CONCURRENTLY по расписанию.

Производительность: реальные числа

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

Оптимизации в порядке impact:

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

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

Метрики которые должны быть с первого дня:

• 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 дней): проектирование. Определяем источники данных, контракты и события, схему БД, требования к latency и объёмам. Прототип ингестора на тестовых данных.

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

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

Фаза 4 (2-3 дня): деплой и мониторинг. Docker Compose или Kubernetes, настройка алертов, runbook.

Итого: 2-4 недели для пайплайна одного протокола. Мультичейн с кросс-чейн агрегацией — 4-8 недель.