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

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

Blockchain — це публічна база даних, у якій можна прочитати все, але зручно прочитати майже нічого. JSON-RPC вузли відповідають на питання "що сталося в блоку X", але не на "покажи мені всі свопи на Uniswap V3 за останні 30 днів за адресами з обсягом > $10k". Для аналітики, risk-систем, торгових стратегій потрібен 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. Blockchain може тимчасово розгалузитися, і останні 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, рекурсивні ліквідації, 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 — стандарт de-facto для аналітичних даних. Компресія 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 та збагачення

Сирі 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, збагачення відбувається по мірі появи 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. Прослуховування Reorganization подій від ноди (WebSocket subscription newHeads + порівняння parentHash). При reorg — видаляємо затроковані блоки зі staging і переприменяємо нову ланцюг.

Для Iceberg це елегантно розв'язується через time travel та merge операції. Для ClickHouse — через ReplacingMergeTree з version стовпцем.

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

-- Normalized events з збагаченням
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 шару (індикатор накопленних reorgs).

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

Етапи розробки