Практика: журнал запусков → дневная витрина

Практика объединяет загрузку файлов, выбор движков, дневную витрину и проверку плана. Используем синтетические данные о завершённых запусках загрузок.

Понадобится Docker-стенд из урока установки. Все команды Docker выполняйте в его папке. Упражнение с четырьмя строками из первой таблицы хранится отдельно и не мешает этой работе.

Файлы и порядок выполнения

Скачайте необходимые файлы в папку стенда:

01-start выполните один раз на новой таблице. Сценарии 02-engines и 04-performance создают собственные таблицы и также рассчитаны на однократное выполнение. Дневную витрину построим пошагово командами этой статьи.

Откройте SQL-файл и выполняйте его блоки в клиенте. Для целого файла подходит следующий способ в Bash и PowerShell:

docker compose cp 01-start.sql ch:/tmp/01-start.sql
docker compose exec -T ch clickhouse-client --user learner --password local-learning-only --multiquery --queries-file /tmp/01-start.sql

Для следующего сценария замените имя файла в обеих командах. Выполнение 02-engines необязательно для pipeline: он создаёт независимые демонстрации движков. Просмотрите его SELECT и объясните выбор версии, сумму и объединение состояния среднего.

Исходные данные

Одна строка runs — один завершённый запуск. Поля: run_id, source, finished_at в UTC, status, rows_loaded и duration_ms. Среднюю длительность считаем по всем запускам, включая неуспешные; количество загруженных строк у failed в этом наборе равно нулю.

После 01-start должно быть 8 запусков, 2 сбоя, 13 700 загруженных строк и 2350 ms суммарной длительности:

SELECT count(), countIf(status = 'failed'),
       sum(rows_loaded), sum(duration_ms)
FROM deshka_ch.runs;

MergeTree не запрещает повтор run_id. Повтор исходного INSERT увеличит итоги; дальнейшие контрольные числа рассчитаны на восемь строк.

Загрузка CSV и JSONL

Чтобы не добавить повтор в runs, создайте две пустые таблицы с её схемой:

CREATE TABLE deshka_ch.import_csv AS deshka_ch.runs;
CREATE TABLE deshka_ch.import_json AS deshka_ch.runs;

Выйдите из клиента. В терминале скопируйте файлы и загрузите каждый в свою таблицу:

docker compose cp runs.csv ch:/tmp/runs.csv
docker compose cp runs.jsonl ch:/tmp/runs.jsonl
docker compose exec -T ch clickhouse-client --user learner --password local-learning-only --query "INSERT INTO deshka_ch.import_csv FROM INFILE '/tmp/runs.csv' FORMAT CSVWithNames"
docker compose exec -T ch clickhouse-client --user learner --password local-learning-only --query "INSERT INTO deshka_ch.import_json FROM INFILE '/tmp/runs.jsonl' FORMAT JSONEachRow"

FROM INFILE читает файл на стороне клиента. Здесь клиент работает в контейнере, поэтому путь /tmp относится к контейнеру. CSV содержит заголовок, JSONL — один объект на строку. На этих файлах оба способа должны дать тот же набор и те же четыре итога, что runs. Повтор команды загрузки снова добавит строки.

Дневная витрина

В клиенте создайте пустую landing-таблицу pipeline_runs и target pipeline_daily. Target хранит число запусков, число сбоев, сумму строк и сумму длительностей; его ключ — день и источник.

CREATE TABLE deshka_ch.pipeline_runs AS deshka_ch.runs;
 
CREATE TABLE deshka_ch.pipeline_daily
(
    day Date,
    source LowCardinality(String),
    runs UInt64,
    failures UInt64,
    rows_loaded UInt64,
    duration_total_ms UInt64
)
ENGINE = SummingMergeTree((runs, failures, rows_loaded, duration_total_ms))
PARTITION BY toYYYYMM(day)
ORDER BY (day, source);
 
CREATE MATERIALIZED VIEW deshka_ch.pipeline_daily_insert
TO deshka_ch.pipeline_daily AS
SELECT toDate(finished_at) AS day, source,
       count() AS runs, countIf(status = 'failed') AS failures,
       sum(rows_loaded) AS rows_loaded,
       sum(duration_ms) AS duration_total_ms
FROM deshka_ch.pipeline_runs
GROUP BY day, source;

View создана до первой вставки. Она агрегирует входной блок, а target хранит частичные суммы. Чтение снова объединяет строки ключа:

INSERT INTO deshka_ch.pipeline_runs SELECT * FROM deshka_ch.runs;
 
SELECT day, source,
       sum(runs) AS total_runs,
       sum(failures) AS total_failures,
       sum(rows_loaded) AS total_rows,
       sum(duration_total_ms) / sum(runs) AS avg_duration_ms
FROM deshka_ch.pipeline_daily
GROUP BY day, source
ORDER BY day, source;

Первый результат имеет шесть групп:

Среднее каждой группы — сумма длительностей, делённая на количество запусков. Общий контроль после первой загрузки — 8, 2, 13700 и 2350. Каждую вставку выполняйте один раз.

Новая порция и сверка

Добавьте два новых события:

INSERT INTO deshka_ch.pipeline_runs VALUES
    (9, 'crm', '2026-01-02 11:00:00', 'ok', 800, 200),
    (10, 'logs', '2026-01-02 11:05:00', 'ok', 4000, 100);

У crm за 2 января станет два запуска, 2000 строк и 350 ms суммарной длительности. У logs — три запуска, один сбой, 10000 строк и 850 ms. Общее среднее всех десяти запусков равно 265 ms.

Сравните два маршрута:

SELECT count() AS runs, countIf(status = 'failed') AS failures,
       sum(rows_loaded) AS rows_loaded, sum(duration_ms) AS duration_total_ms
FROM deshka_ch.pipeline_runs;
 
SELECT sum(runs) AS runs, sum(failures) AS failures,
       sum(rows_loaded) AS rows_loaded, sum(duration_total_ms) AS duration_total_ms
FROM deshka_ch.pipeline_daily;

Оба запроса должны дать 10, 2, 18500, 2650. Дополнительно сверим каждую группу дня и источника: перестановка источников может сохранить общий итог. Запрос ниже складывает разности исходных и сохранённых мер. Ожидается mismatched_groups = 0.

SELECT count() AS mismatched_groups
FROM
(
    SELECT day, source
    FROM
    (
        SELECT toDate(finished_at) AS day, source,
               toInt64(count()) AS runs_delta,
               toInt64(countIf(status = 'failed')) AS failures_delta,
               toInt64(sum(rows_loaded)) AS rows_delta,
               toInt64(sum(duration_ms)) AS duration_delta
        FROM deshka_ch.pipeline_runs
        GROUP BY day, source
 
        UNION ALL
 
        SELECT day, source,
               -toInt64(sum(runs)) AS runs_delta,
               -toInt64(sum(failures)) AS failures_delta,
               -toInt64(sum(rows_loaded)) AS rows_delta,
               -toInt64(sum(duration_total_ms)) AS duration_delta
        FROM deshka_ch.pipeline_daily
        GROUP BY day, source
    )
    GROUP BY day, source
    HAVING sum(runs_delta) != 0 OR sum(failures_delta) != 0
        OR sum(rows_delta) != 0 OR sum(duration_delta) != 0
);

Сравнение ключей на большем наборе

04-performance создаёт perf_by_source и perf_by_id с одинаковыми типами и 200 000 одинаковых строк. Первая отсортирована по (source, finished_at, run_id), вторая — по run_id. Изменяется организация, а не смысл данных.

Сначала сравните EXPLAIN indexes = 1 для суммы duration_ms при source = 'crm'. Посмотрите, сколько гранул осталось. Затем выполните два SELECT с пометками course-source и course-id отдельно несколько раз, чередуя порядок. Проверяйте равенство суммы.

Последние команды файла читают system.query_log: время, read_rows, read_bytes и память. SYSTEM FLUSH LOGS здесь допустим только для собственного учебного сервера; он сбрасывает буфер журнала, а не очищает историю. read_bytes описывает логическое чтение, а не физический объём чтения диска.

Небольшой набор и прогретый кэш могут скрыть разницу времени. Число гранул тоже зависит от состояния частей и настроек. Объясняйте фактические измерения, не ожидайте заранее заданного ускорения.

Объяснить корректность

Повтор события увеличивает обычный счётчик. Поздний merge Replacing-источника не отменяет прежний вклад incremental view. Удаление строки из landing также не уменьшает target автоматически. Эти случаи требуют места дедупликации или согласованного пересчёта периода.

Практика проверяет загрузку и агрегацию на одном сервере. Kafka, реплики и шарды здесь не запущены. Завершайте стенд через docker compose down с сохранением volume, как описано в установке.