Разработка data lake для блокчейн-данных

Разработка data lake для блокчейн-данных

Блокчейн — это публичная база данных, в которой можно прочитать всё, но удобно прочитать почти ничего. JSON-RPC узлы отвечают на вопрос "что произошло в блоке X", но не на "покажи мне все свопы на Uniswap V3 за последние 30 дней по адресам с объёмом > $10k". Для аналитики, риск-систем, торговых стратегий нужен data lake — слой, в котором сырые on-chain данные превращены в структурированные, индексированные, быстро запрашиваемые таблицы.

Источники данных и проблема масштаба

Ethereum mainnet сегодня — это около 20 миллионов блоков, ~2 миллиарда транзакций, терабайты event logs. Полная история Ethereum в формате Parquet занимает ~3–4 TB. На каждый новый блок (~12 секунд) добавляется несколько сотен транзакций и тысячи event logs. BSC, Polygon, Arbitrum, Base — у каждой сети своя история и своя скорость роста.

Три класса данных с разными характеристиками:

Blocks & transactions — структурированные, предсказуемая схема. Главная сложность: reorgs. Блокчейн может временно форкнуться, и последние N блоков могут быть переписаны. Data pipeline должен уметь откатывать уже записанные данные при reorg.

Event logs — самые ценные для аналитики. Transfer, Swap, Liquidation, Mint — всё это EVM events. Проблема: декодирование ABI. Без ABI контракта log — это просто байты с topics. Нужен реестр ABI или сервис вроде Etherscan API / Sourcify для их получения.

Traces (internal transactions) — вызовы между контрактами, которые не создают прямой транзакции. Без трейсов невидима значительная часть DeFi-активности: flash loan внутри одной tx, recursive liquidations, MEV bundle. Получение трейсов через debug_traceTransaction или trace_block — тяжёлая операция, доступна только на archive-узлах.

Архитектура data lake

Слой ingestion

Два подхода к получению данных:

Node-based ingestion — прямое подключение к Ethereum/EVM узлу через WebSocket. Подписка на новые блоки + backfill исторических данных через eth_getLogs batch calls. Требует archive node (или платный доступ к Alchemy/Infura archive). Для backfill миллионов блоков необходима параллельная обработка:

async def fetch_block_range(start: int, end: int, batch_size: int = 100):
    tasks = []
    for i in range(start, end, batch_size):
        chunk = range(i, min(i + batch_size, end))
        tasks.append(fetch_blocks(chunk))
    return await asyncio.gather(*tasks)

Third-party data providers — Goldsky, Envio, Substreams (StreamingFast/Pinax). Они уже проиндексировали историю, вы платите за API или managed pipeline. Быстрее старта, но vendor lock-in и дороже на масштабе.

Хранилище: выбор формата и движка

Для сырых данных блокчейна оптимален columnar storage:

• Apache Parquet на S3/GCS — стандарт де-факто для аналитических данных. Компрессия zstd уменьшает объём event logs в 5–10x. Партиционирование по дате блока + номеру блока.
• Apache Iceberg поверх Parquet — добавляет ACID транзакции, schema evolution, time travel. Критично для обработки reorgs: можно откатить конкретные партиции.
• ClickHouse — OLAP база данных для горячих запросов. Сотни миллионов строк event logs запрашиваются за секунды. Хранит данные columnarly на диске, нет нужды читать Parquet файлы через Spark для каждого запроса.

Типичная двуслойная архитектура:

Raw layer (S3 + Parquet/Iceberg)
    ↓ ETL (dbt / Spark / Flink)
Serving layer (ClickHouse / BigQuery)
    ↓ Query API
Analytics / Trading systems / Dashboards

Декодирование ABI и enrichment

Сырые event logs содержат topics (хэши event signatures) и data (ABI-encoded параметры). Для декодирования нужен ABI реестр:

from eth_abi import decode
from web3 import Web3

TRANSFER_TOPIC = Web3.keccak(text="Transfer(address,address,uint256)").hex()

def decode_transfer(log: dict) -> dict | None:
    if log["topics"][0] != TRANSFER_TOPIC:
        return None
    from_addr = "0x" + log["topics"][1][-40:]
    to_addr   = "0x" + log["topics"][2][-40:]
    amount    = decode(["uint256"], bytes.fromhex(log["data"][2:]))[0]
    return {"from": from_addr, "to": to_addr, "amount": amount}

Для массового декодирования нужен ABI registry — таблица с маппингом contract_address → ABI. Источники: Etherscan API (?module=contract&action=getabi), Sourcify, 4byte.directory для function signatures. Неизвестные контракты обрабатываются как raw bytes, enrichment происходит по мере появления ABI.

Token metadata enrichment: для ERC-20 трансферов нужны decimals, symbol, price at time. Цены берутся из собственных Uniswap V3 TWAP записей или внешних источников (CoinGecko historical API, DeFiLlama).

Обработка reorgs

Reorg — главная головная боль любого блокчейн data pipeline. Когда нода сообщает о новом блоке, у него ещё нет finality. Ethereum с Proof-of-Stake имеет probabilistic finality через несколько блоков и полный finality через ~12.8 минут (2 эпохи). L2 сети имеют ещё более сложную модель.

Стандартный подход:

1. Записывать блоки с confirmation lag (ждать N подтверждений перед записью в "финальный" слой). Для Ethereum: 32–64 блока для практической finality.
2. Хранить "staging" слой для последних M блоков — данные туда пишутся немедленно, но помечаются как pending.
3. Listening на Reorganization события от ноды (WebSocket subscription newHeads + сравнение parentHash). При reorg — удаляем затронутые блоки из staging и переприменяем новую цепочку.

Для Iceberg это элегантно решается через time travel и merge операции. Для ClickHouse — через ReplacingMergeTree с version столбцом.

Схема данных и ключевые таблицы

-- Normalized events с enrichment
CREATE TABLE decoded_events (
    block_number    UInt64,
    block_timestamp DateTime,
    tx_hash         FixedString(66),
    log_index       UInt32,
    contract        FixedString(42),
    event_name      LowCardinality(String),
    chain_id        UInt32,
    -- decoded fields as JSON или отдельные таблицы per event type
    params          String,  -- JSON
    INDEX idx_contract (contract) TYPE bloom_filter GRANULARITY 4,
    INDEX idx_event    (event_name) TYPE set(100) GRANULARITY 4
) ENGINE = ReplacingMergeTree(block_number)
PARTITION BY toYYYYMM(block_timestamp)
ORDER BY (chain_id, contract, block_number, log_index);

Отдельные таблицы для высокочастотных event types: erc20_transfers, uniswap_v3_swaps, aave_liquidations. Партиционирование по месяцам позволяет эффективно запрашивать временные диапазоны.

Инфраструктура и операционные аспекты

Compute: Apache Spark на EMR/Dataproc для batch ETL исторических данных. Apache Flink или Kafka Streams для real-time обработки новых блоков.

Оркестрация: Apache Airflow или Dagster для управления DAG пайплайнов. Отдельные DAG для каждой сети и типа данных.

Мониторинг: lag между последним on-chain блоком и последним записанным — ключевая метрика. Алерт если lag > 5 минут. Также: процент успешно декодированных logs (покрытие ABI реестра), размер staging слоя (индикатор накопленных reorg).

Стоимость архива: полная история Ethereum в ClickHouse — порядка $500–1500/месяц на облачной инфраструктуре в зависимости от провайдера и конфигурации.

Этапы разработки