Разработка системы хранения граббинг-данных (TimescaleDB, ClickHouse)

Разработка системы хранения парсинг-данных (TimescaleDB, ClickHouse)

Сырые данные из блокчейна или бирж накапливаются быстро — десятки гигабайт в день для активно парсимых источников. Хранить это в обычном PostgreSQL в одной таблице означает деградацию запросов через несколько месяцев. Выбор между TimescaleDB и ClickHouse — это выбор между двумя принципиально разными моделями хранения и разными паттернами запросов. Обе технологии правильные в своём контексте.

TimescaleDB vs ClickHouse: когда что

TimescaleDB — расширение PostgreSQL. Добавляет hypertables (автоматическое партиционирование по времени), continuous aggregates (инкрементальные материализованные представления), compression. Вы остаётесь в PostgreSQL-экосистеме: стандартный SQL, ACID транзакции, JOIN с обычными таблицами, привычный инструментарий.

ClickHouse — колоночная OLAP база данных. Данные хранятся по столбцам, что даёт огромный выигрыш при агрегациях по подмножеству колонок. Скорость GROUP BY и SUM на миллиардах строк — на порядок выше PostgreSQL. Слабая сторона: нет транзакций, UPDATE/DELETE — дорогие операции, JOIN работает иначе.

Рекомендация для парсинга on-chain данных:

• TimescaleDB — если данные нужны для продуктовой логики (балансы, позиции, аккаунты), есть JOIN с реляционными данными, нужны ACID-гарантии
• ClickHouse — если это аналитический pipeline (trading сигналы, агрегированная статистика, исторический анализ), запросы работают с большими диапазонами дат

В production часто комбинируют: TimescaleDB для горячих/операционных данных + ClickHouse для аналитического warehouse.

Архитектура TimescaleDB

Hypertables

Базовая концепция: обычная таблица PostgreSQL превращается в hypertable — под капотом создаются чанки (partitions) по временному измерению. Каждый чанк — отдельный файл, старые чанки можно компрессировать или архивировать.

-- Создаём таблицу
CREATE TABLE trades (
  time        TIMESTAMPTZ NOT NULL,
  exchange    TEXT NOT NULL,
  symbol      TEXT NOT NULL,
  price       NUMERIC(20, 8) NOT NULL,
  volume      NUMERIC(20, 8) NOT NULL,
  side        CHAR(4) NOT NULL  -- 'buy' | 'sell'
);

-- Превращаем в hypertable с чанками по 1 дню
SELECT create_hypertable('trades', 'time', chunk_time_interval => INTERVAL '1 day');

-- Индексы per chunk (TimescaleDB создаёт автоматически на time)
CREATE INDEX ON trades (symbol, time DESC);

Размер чанка — ключевой параметр. Правило: каждый чанк должен помещаться в памяти для bulk insert (обычно 1-7 дней для биржевых данных).

Continuous Aggregates

Заменяют дорогие realtime GROUP BY на инкрементальные материализованные представления:

-- Агрегат OHLCV по минутам
CREATE MATERIALIZED VIEW trades_1m
WITH (timescaledb.continuous) AS
SELECT
  time_bucket('1 minute', time) AS bucket,
  symbol,
  exchange,
  first(price, time)  AS open,
  max(price)          AS high,
  min(price)          AS low,
  last(price, time)   AS close,
  sum(volume)         AS volume,
  count(*)            AS trade_count
FROM trades
GROUP BY bucket, symbol, exchange
WITH NO DATA;

-- Политика обновления: обновлять агрегат каждые 1 минуту, начиная с 2 минут назад
SELECT add_continuous_aggregate_policy('trades_1m',
  start_offset => INTERVAL '2 minutes',
  end_offset   => INTERVAL '1 minute',
  schedule_interval => INTERVAL '1 minute'
);

Теперь запрос SELECT * FROM trades_1m WHERE bucket > NOW() - INTERVAL '1 day' — это SELECT из материализованного представления, не агрегация по raw данным.

Compression

Старые данные компрессируются почти без потери функциональности (кроме UPDATE/DELETE):

-- Включаем компрессию с сортировкой по symbol+time (оптимально для наших запросов)
ALTER TABLE trades SET (
  timescaledb.compress,
  timescaledb.compress_orderby = 'time DESC',
  timescaledb.compress_segmentby = 'symbol'
);

-- Автоматически компрессировать чанки старше 7 дней
SELECT add_compression_policy('trades', INTERVAL '7 days');

Типичная степень сжатия для биржевых данных: 10-20x. 100GB raw → 5-10GB compressed.

Retention policy

-- Автоматически удалять данные старше 2 лет
SELECT add_retention_policy('trades', INTERVAL '2 years');

-- Или: переместить старые чанки на дешёвое хранилище (tablespace на HDD)
SELECT add_tiering_policy('trades', INTERVAL '30 days');

Архитектура ClickHouse

MergeTree engines

Выбор engine — критический момент. Для парсинг-данных чаще всего используются:

MergeTree — базовый, без специальных свойств:

CREATE TABLE trades
(
    time      DateTime64(3),
    exchange  LowCardinality(String),
    symbol    LowCardinality(String),
    price     Decimal(20, 8),
    volume    Decimal(20, 8),
    side      Enum8('buy' = 1, 'sell' = 2)
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(time)
ORDER BY (symbol, exchange, time);

ORDER BY в ClickHouse — это одновременно и первичный ключ (sparse index), и физический порядок хранения. Выбирайте по паттернам запросов: если чаще фильтруете по (symbol, time) — именно такой ORDER BY.

ReplacingMergeTree — для дедупликации при insert (нужно для надёжного re-processing):

CREATE TABLE balances
(
    time      DateTime64(3),
    address   String,
    token     LowCardinality(String),
    balance   Decimal(38, 18),
    block_number UInt64,
    _version  UInt64  -- обычно block_number или timestamp для dedup
)
ENGINE = ReplacingMergeTree(_version)
PARTITION BY toYYYYMM(time)
ORDER BY (address, token, time);

При дублирующихся вставках (same primary key) — сохраняется запись с максимальным _version. Deduplication происходит при merge, не сразу — для точных запросов нужен FINAL modifier.

SummingMergeTree — для агрегатов (объёмы, суммы):

CREATE TABLE volume_by_hour
(
    hour     DateTime,
    symbol   LowCardinality(String),
    volume   Decimal(20, 8),
    count    UInt64
)
ENGINE = SummingMergeTree()
PARTITION BY toYYYYMM(hour)
ORDER BY (symbol, hour);

Batch insert

ClickHouse оптимизирован для вставки батчами. Частые мелкие вставки — анти-паттерн:

import clickhouse_connect

client = clickhouse_connect.get_client(host='localhost')

# Правильно: батч минимум 1000-10000 строк
def flush_buffer(rows: list[dict]):
    if len(rows) < 1000:
        return  # накапливаем дальше
    
    client.insert(
        'trades',
        [[r['time'], r['exchange'], r['symbol'], r['price'], r['volume']] for r in rows],
        column_names=['time', 'exchange', 'symbol', 'price', 'volume']
    )
    rows.clear()

Минимальный рекомендуемый батч: 1000 строк. Оптимальный: 10,000-100,000 строк. Частота: не чаще 1-2 раз в секунду на таблицу.

Materialized Views в ClickHouse

ClickHouse materialized views — триггерные, обновляются при insert (не по расписанию как в TimescaleDB):

-- Materialized view для агрегации OHLCV
CREATE MATERIALIZED VIEW trades_1h_mv
TO trades_1h  -- целевая таблица
AS
SELECT
    toStartOfHour(time) AS hour,
    symbol,
    exchange,
    argMin(price, time)  AS open,
    max(price)           AS high,
    min(price)           AS low,
    argMax(price, time)  AS close,
    sum(volume)          AS volume
FROM trades
GROUP BY hour, symbol, exchange;

ASOF JOIN для временных данных

Уникальная функция ClickHouse для временных данных — джойн по ближайшему значению:

-- Присоединить цену актива к каждой ликвидации (ближайшая цена до события)
SELECT 
    l.time,
    l.symbol,
    l.quantity_usd,
    p.price AS price_at_liquidation
FROM liquidations l
ASOF LEFT JOIN prices p
    ON l.symbol = p.symbol AND l.time >= p.time;

Общие практики

Типы данных. Используйте LowCardinality(String) для полей с малой кардинальностью (exchange, symbol, side) — экономия 2-10x по размеру и ускорение фильтрации. Decimal вместо Float для финансовых значений — нет проблем с точностью.

Партиционирование. По месяцам (toYYYYMM) — стандарт для большинства финансовых данных. Позволяет дропать старые партиции без DELETE.

Мониторинг. Ключевые метрики:

• Размер таблиц и партиций
• Количество parts (в ClickHouse: много маленьких parts = проблема)
• Время выполнения медленных запросов
• Insert latency при пиковой нагрузке

Стек

Проектирование схемы + настройка TimescaleDB или ClickHouse для конкретного объёма и паттернов запросов: 1-2 недели. Если включает миграцию существующих данных — плюс неделя.