Стратегии дедупликации данных

Дедупликация — фундаментальная задача в ClickHouse. Из-за append-only семантики MergeTree (INSERT никогда не заменяет существующие строки) дублирующиеся записи неизбежны: повторные доставки из Kafka, CDC-репликация, ETL retry.

В ClickHouse существует четыре стратегии дедупликации. Каждая — trade-off между производительностью чтения, стоимостью записи, и гарантиями консистентности.

Стратегия 1: ReplacingMergeTree + FINAL

Самый простой и распространённый подход. ReplacingMergeTree хранит все версии записи, а модификатор FINAL обеспечивает дедупликацию при чтении:

CREATE TABLE orders (
    order_id UInt64,
    status LowCardinality(String),
    amount Decimal64(2),
    updated_at DateTime
) ENGINE = ReplacingMergeTree(updated_at)
ORDER BY order_id;

-- Две версии одного заказа
INSERT INTO orders VALUES (1001, 'created', 99.99, '2024-01-15 10:00:00');
INSERT INTO orders VALUES (1001, 'shipped', 99.99, '2024-01-15 14:00:00');

-- Дедуплицированное чтение
SELECT order_id, status, amount
FROM orders FINAL
WHERE order_id = 1001;
-- Результат: (1001, 'shipped', 99.99)

Вертикальный FINAL (24.x+)

Начиная с ClickHouse 24.x, FINAL значительно оптимизирован. “Вертикальный FINAL” (vertical FINAL) обрабатывает только столбцы, упомянутые в запросе, вместо всей строки. Это снижает I/O в запросах, читающих подмножество столбцов:

-- Читает только order_id + status (2 столбца вместо 4)
SELECT order_id, status FROM orders FINAL WHERE order_id = 1001;

Когда использовать

• Dimension-таблицы (клиенты, продукты, локации) — SCD Type 1
• Небольшие/средние таблицы (миллионы строк)
• Простота: минимальный ETL, одна таблица

Стратегия 2: argMax / argMin (query-time дедупликация)

argMax/argMin — дедупликация через GROUP BY в SELECT. Не требует специального engine:

-- Работает с любым MergeTree (не только ReplacingMergeTree)
SELECT
    order_id,
    argMax(status, updated_at) AS status,
    argMax(amount, updated_at) AS amount
FROM orders
GROUP BY order_id;

argMax(value, version) возвращает value из строки с максимальным version. Эквивалент FINAL, но реализованный на уровне запроса.

Когда argMax лучше FINAL

argMax выигрывает у FINAL, когда фильтр отсекает менее 50% данных. FINAL обрабатывает все parts, содержащие совпадающие ORDER BY ключи. argMax работает с уже отфильтрованным набором данных:

-- Сценарий: 90% заказов в статусе 'shipped'
-- FINAL обрабатывает все parts, затем фильтрует
SELECT * FROM orders FINAL WHERE status = 'shipped';

-- argMax: сначала GROUP BY (дедупликация), затем HAVING
SELECT order_id, argMax(status, updated_at) AS status
FROM orders
GROUP BY order_id
HAVING status = 'shipped';

Когда FINAL лучше argMax

• Point lookup по ORDER BY ключу (WHERE order_id = X) — FINAL может использовать primary key для прямого обращения к parts
• Запросы без GROUP BY — argMax требует явного GROUP BY по всем ключам

Стратегия 3: Deduplicated MV (дедупликация при INSERT)

Самая производительная стратегия для чтения. Materialized View с AggregatingMergeTree + argMaxState дедуплицирует данные при вставке, а не при чтении:

-- Исходная таблица (append-only, все версии)
CREATE TABLE raw_orders (
    order_id UInt64,
    status String,
    amount Decimal64(2),
    updated_at DateTime
) ENGINE = MergeTree()
ORDER BY (order_id, updated_at);

-- Target-таблица (дедуплицированная)
CREATE TABLE deduped_orders (
    order_id UInt64,
    status AggregateFunction(argMax, String, DateTime),
    amount AggregateFunction(argMax, Decimal64(2), DateTime)
) ENGINE = AggregatingMergeTree()
ORDER BY order_id;

-- MV: при каждом INSERT в raw_orders автоматически дедуплицирует
CREATE MATERIALIZED VIEW orders_dedup_mv TO deduped_orders AS
SELECT
    order_id,
    argMaxState(status, updated_at) AS status,
    argMaxState(amount, updated_at) AS amount
FROM raw_orders
GROUP BY order_id;

Чтение из target-таблицы — без FINAL, без GROUP BY:

SELECT
    order_id,
    argMaxMerge(status) AS status,
    argMaxMerge(amount) AS amount
FROM deduped_orders
GROUP BY order_id;

Зачем нужен argMaxMerge?

Столбцы типа AggregateFunction(argMax, ...) хранят бинарное промежуточное состояние. argMaxMerge() финализирует это состояние в конечное значение. Прямой SELECT без argMaxMerge вернёт бессмысленный blob.

Стратегия 4: OPTIMIZE TABLE FINAL (batch cleanup)

OPTIMIZE TABLE FINAL — принудительное физическое слияние всех parts:

-- Принудительный merge всех parts
OPTIMIZE TABLE orders FINAL;

-- После OPTIMIZE обычный SELECT (без FINAL) возвращает корректный результат
SELECT * FROM orders WHERE order_id = 1001;

Почему это стратегия дедупликации?

После OPTIMIZE TABLE FINAL обычный SELECT эквивалентен SELECT FINAL — потому что все parts уже объединены и дедуплицированы. Это позволяет использовать ReplacingMergeTree без overhead FINAL на каждый запрос.

Anti-pattern: OPTIMIZE TABLE FINAL в application code

-- НИКОГДА не делайте это в production:
-- 1. INSERT INTO orders VALUES (...)
-- 2. OPTIMIZE TABLE orders FINAL  -- O(all data)!
-- 3. SELECT * FROM orders WHERE ...

OPTIMIZE TABLE FINAL — операция O(all data). Она переписывает все parts таблицы, независимо от количества новых данных. На таблице с миллиардами строк — это часы I/O и полная нагрузка на диск.

Comparison matrix: 4 стратегии

Performance matrix

Decision tree

1. Dimension-таблица до 100M строк? — ReplacingMergeTree + FINAL (простота)
2. High-throughput pipeline (1M+ строк/с), критичная скорость чтения? — Deduplicated MV
3. Аналитические запросы с GROUP BY, фильтрация менее 50%? — argMax/argMin
4. Ночной batch cleanup для dashboards? — OPTIMIZE TABLE FINAL по расписанию

Полный пример: CDC pipeline с дедупликацией

-- Source: PostgreSQL orders через Debezium CDC
-- Каждое изменение статуса -> INSERT в ClickHouse

-- Raw-таблица (все версии)
CREATE TABLE raw_orders (
    order_id UInt64,
    customer_id UInt64,
    status LowCardinality(String),
    amount Decimal64(2),
    updated_at DateTime
) ENGINE = MergeTree()
ORDER BY (customer_id, order_id, updated_at);

-- Дедуплицированный view для аналитики
CREATE TABLE orders_latest (
    order_id UInt64,
    customer_id UInt64,
    status AggregateFunction(argMax, LowCardinality(String), DateTime),
    amount AggregateFunction(argMax, Decimal64(2), DateTime)
) ENGINE = AggregatingMergeTree()
ORDER BY (customer_id, order_id);

CREATE MATERIALIZED VIEW orders_dedup_mv TO orders_latest AS
SELECT
    order_id,
    customer_id,
    argMaxState(status, updated_at) AS status,
    argMaxState(amount, updated_at) AS amount
FROM raw_orders
GROUP BY order_id, customer_id;

-- Аналитический запрос: revenue по статусам
SELECT
    argMaxMerge(status) AS status,
    count() AS orders,
    sum(argMaxMerge(amount)) AS revenue
FROM orders_latest
GROUP BY order_id
HAVING status = 'paid';

Ключевые выводы

1. ReplacingMergeTree + FINAL — простейший паттерн. Одна таблица, FINAL при чтении. Для dimension-таблиц и SCD.
2. argMax/argMin — query-time деdup через GROUP BY. Лучше FINAL при фильтрации менее 50% данных.
3. Deduplicated MV — максимальная производительность чтения. INSERT-time деdup через argMaxState + AggregatingMergeTree.
4. OPTIMIZE TABLE FINAL — batch cleanup. O(all data). Не для production reads.
5. Anti-pattern: OPTIMIZE TABLE FINAL в application code. Это batch-операция для maintenance, не для runtime дедупликации.