Партиционированные витрины и инкрементальные апдейты через MERGE

До этого момента витрина mart_daily_zone жила как VIEW — пересчитывалась от сырых файлов на каждый запрос. Для разработки это удобно, но в работе так нельзя: гонять 41 миллион поездок при каждом обращении аналитика расточительно. Витрину нужно материализовать — сохранить посчитанный результат.

И здесь возникает главный вопрос капстоуна: когда приходит новый месячный файл, как обновить витрину, не пересчитывая всю историю? Ответ — материализовать витрину в DuckLake, партиционировать её по месяцам и обновлять инкрементально через MERGE INTO. Этот урок собирает воедино DuckLake (модуль про лейкхаусы) и MERGE (модуль про запись данных) на конкретной задаче.

Материализация витрины в DuckLake

Перенесём витрину из VIEW в физическую таблицу DuckLake-лейкхауса, который мы подключили в первом уроке модуля. Это та же CREATE TABLE AS, но целевая база — лейкхаус lake:

-- Материализуем витрину как таблицу в DuckLake-лейкхаусе
CREATE TABLE lake.mart_daily_zone AS
FROM mart_daily_zone;     -- mart_daily_zone — VIEW из урока про ELT

-- Результат:
-- 287_044 строк записано в лейкхаус.
-- Данные легли Parquet-файлами в capstone_lake/,
-- метаданные и первый снапшот — в SQLite-каталог.

Теперь lake.mart_daily_zone — это материализованная таблица: данные в Parquet на объектном хранилище, метаданные в SQL-каталоге. Запрос аналитика к ней читает готовый результат, а не пересчитывает 41 миллион поездок. И, поскольку это DuckLake, у витрины бесплатно появились ACID, снапшоты и time-travel — каждое будущее обновление будет атомарным и версионированным.

Партиционирование по месяцам

Просто материализовать витрину мало — её нужно правильно физически разложить. Витрина естественно делится по времени: данные за апрель 2026 не пересекаются с данными за март. Это и есть основа для партиционирования.

Партиционирование — это раскладка данных таблицы по отдельным группам файлов по значению ключа. Если разложить витрину по (year, month), то все строки за конкретный месяц лежат в своей группе файлов отдельно от других месяцев.

-- Витрина, партиционированная по году и месяцу
CREATE TABLE lake.mart_daily_zone (
    trip_date DATE, borough VARCHAR, zone_name VARCHAR,
    trips BIGINT, revenue DECIMAL(18,2),
    avg_distance_mi DOUBLE, avg_duration_min DOUBLE,
    year INTEGER, month INTEGER
);

-- Ключ партиционирования — год и месяц
ALTER TABLE lake.mart_daily_zone SET PARTITIONED BY (year, month);

-- Первоначальная загрузка истории
INSERT INTO lake.mart_daily_zone FROM mart_daily_zone;

Партиционирование даёт два эффекта, и оба важны для капстоуна.

Первый эффект — partition pruning при чтении. Запрос с фильтром WHERE year = 2026 AND month = 4 читает только файлы партиции апреля, а остальные 27 месяцев физически не трогает. Это сужение объёма чтения ещё до zonemaps.

Второй эффект, ключевой для капстоуна, — локальность обновления. Новый месячный файл порождает данные только за этот месяц. Раз каждый месяц — своя партиция, обновление затрагивает одну партицию, а 27 партиций истории остаются нетронутыми. Это и есть фундамент инкрементальности.

Инкрементальный апдейт: проблема и инструмент

Пришёл новый файл raw/trips/year=2026/month=05/trips.parquet. Наивный путь — CREATE OR REPLACE TABLE всей витрины из всех сырых файлов — означает пересчёт 41 миллиона строк ради одного месяца. Это то, чего инкрементальность должна избегать.

Правильный путь: посчитать агрегаты только за новый месяц и влить их в витрину. Но «влить» — это не всегда чистый INSERT. Возможна ситуация, когда часть месяца уже была загружена (поздно дошедшие записи, перезаливка частичного файла), и тогда нужно не продублировать строки, а обновить существующие и добавить недостающие. Именно для этого существует MERGE INTO.

MERGE INTO (появился в DuckDB 1.4) — это операция «обнови или вставь» одним выражением. Она сопоставляет целевую таблицу с источником по условию и для совпавших строк делает UPDATE, для несовпавших — INSERT. В отличие от INSERT ... ON CONFLICT, MERGE не требует первичного ключа и работает по произвольному условию слияния.

MERGE на практике

Соберём инкрементальный апдейт целиком. Сначала — staging и агрегаты только за новый месяц (то есть наш ELT-пайплайн из урока 2, но с фильтром на один файл):

-- Агрегаты только за новый месяц — источник для MERGE
CREATE OR REPLACE VIEW mart_new_month AS
SELECT
    t.pickup_at::DATE AS trip_date, z.borough, z.zone_name,
    count(*) AS trips, sum(t.total_amount) AS revenue,
    avg(t.distance_mi) AS avg_distance_mi,
    avg(epoch(t.dropoff_at - t.pickup_at) / 60) AS avg_duration_min,
    t.year, t.month,
FROM read_parquet('s3://raw/trips/year=2026/month=05/trips.parquet',
                   hive_partitioning = true) AS t
JOIN stg_zones AS z ON t.pu_location_id::INTEGER = z.zone_id
WHERE t.total_amount >= 0 AND t.dropoff_datetime > t.pickup_datetime
GROUP BY ALL;

Теперь — сам MERGE. Целевая таблица — витрина в лейкхаусе, источник — агрегаты нового месяца, ключ слияния — (trip_date, borough, zone_name):

-- Инкрементальный апдейт витрины: обновить совпавшее, вставить новое
MERGE INTO lake.mart_daily_zone AS target
USING mart_new_month AS source
ON  target.trip_date = source.trip_date
AND target.borough   = source.borough
AND target.zone_name = source.zone_name
WHEN MATCHED THEN UPDATE SET
    trips = source.trips, revenue = source.revenue,
    avg_distance_mi = source.avg_distance_mi,
    avg_duration_min = source.avg_duration_min
WHEN NOT MATCHED THEN INSERT VALUES (
    source.trip_date, source.borough, source.zone_name,
    source.trips, source.revenue, source.avg_distance_mi,
    source.avg_duration_min, source.year, source.month
);

-- Результат:
-- обработано 9_812 строк нового месяца:
-- часть пошла в UPDATE (день/зона уже были), часть — в INSERT.
-- 27 партиций истории не тронуты.

Что важно в этой связке. Объём работы — один месяц (около 10 тысяч строк агрегатов и один сырой Parquet-файл), а не вся история. Затронута одна партиция витрины. Операция идемпотентна: повторный запуск того же MERGE с тем же источником снова попадёт в WHEN MATCHED, перезапишет строки теми же значениями и не создаст дублей — пайплайн можно безопасно перезапускать. И поскольку витрина в DuckLake, весь MERGE атомарен: либо все изменения видны, либо ни одного, а сам апдейт фиксируется новым снапшотом.

Снапшот и проверка

После MERGE в DuckLake появился новый снапшот витрины. Это даёт встроенную проверку и возможность отката:

-- История витрины: каждый MERGE — новый снапшот
FROM lake.snapshots();
-- snapshot_id | snapshot_time       | schema_version
-- 0           | 2026-05-20 10:01:11 | 1
-- 1           | 2026-05-20 10:42:55 | 1   <- результат MERGE нового месяца

-- Сравнить витрину до и после апдейта через time-travel
SELECT
    (SELECT count(*) FROM lake.mart_daily_zone AT (VERSION => 0)) AS before,
    (SELECT count(*) FROM lake.mart_daily_zone) AS after_merge;
-- before  | after_merge
-- 287_044 | 296_856

Time-travel из урока про DuckLake здесь работает как инструмент эксплуатации: можно сверить состояние витрины до и после загрузки, а если новый месяц оказался битым — запросить или восстановить предыдущий снапшот. Инкрементальный апдейт получается не только быстрым, но и обратимым.

Попробуй сам

Понадобится DuckDB 1.5.x с расширением ducklake и витрина из предыдущих уроков модуля.

Задания:

1. Материализуйте витрину в DuckLake-лейкхаусе через CREATE TABLE lake.mart_daily_zone AS .... Затем задайте партиционирование по (year, month) и посмотрите файловую раскладку в папке данных — появились ли отдельные группы файлов по месяцам.
2. Возьмите данные одного «нового» месяца (отдельный Parquet или подвыборку), посчитайте по ним агрегаты-источник и выполните MERGE INTO в витрину с ключом (trip_date, borough, zone_name).
3. Запустите тот же MERGE второй раз без изменений. Убедитесь, что число строк витрины не выросло — это и есть идемпотентность.
4. Выполните FROM lake.snapshots(); и сравните count(*) витрины на снапшоте до MERGE и после через AT (VERSION => N). Сформулируйте, как time-travel помогает откатить неудачную загрузку.