On-chain ETL-пайплайн: разработка для блокчейн-данных

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

Пример из практики: один из наших клиентов потерял 2 недели на восстановление данных из-за неправильной обработки реоргов. После внедрения нашего пайплайна экономия времени составила 40% на исторической синхронизации, а стоимость владения инфраструктурой снизилась на 30% за счёт оптимизации хранения.

ETL без обработки реорганизаций — это не ETL, а генерация мусора. Наш опыт говорит: 90% проблем решаются правильной архитектурой на старте.

Что такое on-chain ETL-пайплайн и зачем он нужен?

On-chain ETL-пайплайн — это система, которая извлекает сырые данные из блокчейна (логи событий, внутренние транзакции, изменения стейта), трансформирует их в структурированные записи (декодирование ABI, обогащение ценами, нормализация сумм) и загружает в аналитическое хранилище. Без такого пайплайна невозможно строить дашборды DeFi-протоколов, отслеживать ликвидность в реальном времени или проводить исторический анализ. Основные сложности: реорганизации цепочек, огромные объёмы (до 15M+ блоков в Ethereum) и необходимость гарантировать консистентность при параллельной ингестии.

Как устроена архитектура: три слоя 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 ускорение ингестии.

Экономия до 40% времени на исторической синхронизации за счёт параллельной ингестии — проверено на проектах с нагрузкой 10 000 событий/сек.

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

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

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

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

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

Фаза 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 недель. Стоимость разработки рассчитывается индивидуально. Аудит существующего пайплайна — по запросу. Получите консультацию по вашему проекту — свяжитесь с нами для бесплатной оценки.

Что входит в работу

• Документация архитектуры и схемы данных
• Исходный код пайплайна (GitHub)
• Настроенный мониторинг (Grafana дашборды, Prometheus алерты)
• Инструкция по деплою и эксплуатации
• Обучение команды (2-3 сессии)
• Поддержка в течение 1 месяца после запуска

Наша команда имеет 8+ лет опыта в блокчейн-разработке и более 50 завершённых ETL-проектов. Мы используем только проверенные инструменты и гарантируем надёжность пайплайна даже при пиковых нагрузках. Закажите разработку или аудит — получите готовое решение в сжатые сроки.