clickstream-data-platform/docs/architecture/storage.md

Хранилище: слои и конвенции

Документ описывает сторону ClickHouse: как называются объекты, какие служебные колонки у них общие, чем нарезаны и сколько живут данные, как устроен приём и из каких файлов собирается DDL. Здесь же карта таблиц, которая растёт с этапами.

Что здесь описано и чего ещё нет. Собран этап 1: кластер из двух шардов, keeper, Kafka, каркас сервисов. Объекты хранилища и механизм применения DDL закладывает этап 2 — на момент написания их в репозитории нет. Дальше по тексту устройство описано так, как оно проектируется; построенное от заложенного отличает карта таблиц в конце.

Зона ответственности у документа одна — хранилище. Генератор описан отдельно: его замысел — в спеке генератора, формат события — в описании выгрузки. Хранилище строится по описанию выгрузки, как в бою строится по документации источника. Целевая картина всего стенда — спека «Боевой реализм стенда (v2)».

Имена объектов

Имя объекта заканчивается тем, что это за объект:

Суффикс	Что это
_rep	локальная таблица шарда, движок семейства Replicated*
_dist	Distributed поверх одноимённой локальной
_kafka	таблица на движке Kafka
_mv	материализованное представление
_v	обычное представление

Суффикс носит каждый физический объект. Голого имени у таблицы не существует: запрос к stg.hits_raw даёт громкую ошибку «нет такой таблицы» — а под голым именем в документах и разговоре понимается сущность, у которой этих объектов несколько. Единственное исключение — словари: у них воплощение одно, шардировать нечего, и суффикс ничего не различал бы.

Распространённая конвенция, где голое имя означает локальную таблицу, а распределённая получает суффикс _all, ошибается иначе: забытый суффикс тихо возвращает данные одного шарда. Правило спеки «проверки и контрольные суммы — только по Distributed» такую тишину переживает плохо.

Второе свойство — SHOW TABLES группирует объекты по сущности, а не по технологии: все четыре объекта топика hits стоят рядом, потому что различаются хвостом, а не началом имени.

Раскладка по шардам

Пишем только в _dist. Локальные таблицы остаются для чтения и обслуживания — операций с партициями, ручной переобработки. Правило не про удобство: при записи через распределённую таблицу раскладку определяет ключ шардирования, то есть свойство данных, а при записи в локальную — то, какая нода случайно выполняла код. Отсюда урок стенда: какая нода читала топик, меняется между прогонами (видно в колонке consumer_host), а куда легли данные — нет.

Ключи шардирования: cityHash64(ClientID) у событий, cityHash64(order_id) у заказов, cityHash64 сырой строки у STG и у таблицы ошибок. У первых двух хеш выбран против перекоса: структурированный числовой идентификатор распределяется по остатку от деления неравномерно. У сырья выбора нет — строку иначе не разложишь; там хеш даёт другое свойство, одинаковые сообщения ложатся на один шард.

Ключи у слоёв разные, и это имеет наблюдаемое следствие: сырая строка и разобранное из неё событие почти всегда оказываются на разных шардах. Пошардовые счётчики STG и ODS поэтому не сходятся и сходиться не должны — сверять слои можно только через _dist.

Служебные колонки

Собственные колонки не повторяют имён виртуальных. Виртуальные даёт движок: _topic, _partition, _offset, _timestamp у Kafka, _shard_num у Distributed и прочие. Если положить на диск колонку с таким же именем, в матвью перестанет читаться, что дано движком, а что положено нами, — а это ровно то различие, ради которого метаданные доставки и хранятся. Поэтому они ложатся под именами kafka_topic, kafka_partition, kafka_offset, kafka_timestamp, рядом — consumer_host, имя читавшей ноды: виртуальные колонки его не несут, а после записи в Distributed он уже невосстановим.

Само сообщение лежит в колонке raw тем, чем пришло: чтец читает байты и ничего не проверяет, поэтому там оказываются и целые события, и мусор. Разбирается всё это ниже, в матвью ODS — см. ADR 0005.

Метка времени загрузки зовётся _load_ts, тип DateTime64(3). В ODS она же служит колонкой версии ReplacingMergeTree. Имя согласовано с каноном служебных полей соседнего учебного стенда на Greenplum, чтобы словарь был общим у двух хранилищ; ведущее подчёркивание у технических колонок — распространённая запись, её же используют Fivetran, Airbyte и Stitch. С правилом выше это не спорит: запрещено совпадать с именами виртуальных колонок, а не носить подчёркивание.

Идентификатора пачки загрузки (_load_id) пока нет. В STG и ODS данные приезжают потоком через матвью, у которого нет ни батча, ни run_id, и колонка была бы пустой формальностью. В слоях, которые наполняет Airflow, run_id появится по-настоящему — тогда и заведём, тем же стилем имени.

Путь реплицированных таблиц в keeper

Шаблон — /clickhouse/tables/{shard}/{database}/{table}. База в пути обязательна: без неё одноимённые таблицы разных слоёв получат один и тот же узел в keeper и подерутся. Макрос {uuid} не используем, хотя он тоже развёл бы пути: он завязан на движок базы Atomic и делает путь нечитаемым, а на учебном стенде возможность открыть system.zookeeper и увидеть осмысленный путь — сама по себе половина урока про то, чем занят keeper.

Приём: поток и его свойства

Цепочка одна: чтец топика → матвью → сырьё STG → матвью разбора → событие и таблица ошибок ODS.

Источник матвью разбора — stg.hits_raw_dist, а не локальная таблица. У матвью две привязки: источник, на вставку в который она срабатывает, и цель, куда пишет. Распределённая таблица — лицо слоя, локальная — его хранилище; потребитель слоя цепляется к лицу. Практически это значит, что разбор идёт на той же ноде, что читала Kafka, в момент вставки первой матвью — до раскладки по шардам.

Вставка синхронная. На пути приёма стоит distributed_foreground_insert = 1. По умолчанию вставка в распределённую таблицу кладёт блок в локальный спул и сразу возвращает управление, а Kafka коммитит офсеты по факту работы матвью — то есть по факту записи в спул. Топик уже считает сообщение прочитанным, хотя на шарде его нет. Синхронный режим не добавляет работы, он её не прячет: кусок на шарде будет записан всё равно, вопрос лишь в том, ждём ли мы этого внутри вставки. Платится ожидание один раз на блок Kafka в десятки тысяч строк, а не на событие.

Гарантия — «хотя бы один раз», не транзакция. Падение после записи на шард, но до коммита офсетов даёт повтор при перечитывании. В ODS повтор схлопнет ReplacingMergeTree, а сырьё дедупа не имеет вовсе: перезаливка модельного дня честно удваивает count() в STG, и живёт эта пара до истечения срока хранения. Это свойство слоя, а не поломка, — но обещание идемпотентности конвейера к сырому слою не относится.

Матвью разбора две, и их условия обязаны делить поток без зазора и без нахлёста. Одна забирает годные строки в ods.event_dist, вторая — брак в ods.event_errors_dist. Строка, подошедшая обеим, задвоится; не подошедшая ни одной — исчезнет молча. Держится это формой: второе условие пишется буквальным отрицанием первого, а сам предикат собирается только из функций, не возвращающих NULL, — иначе трёхзначная логика даст строку, которую не возьмёт ни условие, ни NOT условие.

Постоянной сверки счётчиков при этом нет и не должно быть. У сырья срок жизни трое суток, а ODS хранит всё, поэтому равенство «сырьё = события + ошибки» разъедется само; переобработка добавит строк в ODS, перезаливка задвоит сырьё, а ODS её схлопнет. Правило, красное в норме, учит не смотреть на оповещения (ADR 0002). Равенство проверяется разово в smoke на управляемой пачке: отправили N сообщений — получили N строк сырья и N в сумме событий и ошибок.

Срок жизни сырья

Сырьё в STG живёт трое суток реального времени и уходит само. Трое — это окно отладки: столько сырьё лежит на ноутбуке, чтобы менти успел разобрать полёты, после чего перестаёт занимать место. Нарезка — по дню загрузки, срок — по той же колонке _load_ts, снятие — целыми кусками (ttl_only_drop_parts). Значение этой настройки между версиями ClickHouse менялось, поэтому в DDL оно проставлено явно.

Нарезать сырьё по модельному дню события было бы соблазнительно — он единица переобработки и он же ключ партиции в ODS, — но чистку это ломает. Кусок снимается целиком, только когда в нём истекли все строки, а обычный мерж внутри партиции модельного дня склеит куски разного возраста, и данные переживут срок. В партиции дня загрузки склеивать нечего: все строки в ней одного возраста, и куски уходят по мере того, как истекает самый свежий из них — то есть сырьё живёт трое суток с небольшим хвостом, а не ровно трое.

Модельного дня среди колонок сырья нет вовсе. Он свойство содержимого, а содержимое разбирает ODS — там EventDate и живёт, ключом партиции. Сырьё режется своими координатами: и разбор полётов, и переобработка фильтруют по _load_ts, попадая в ключ партиции, а не идя сплошным проходом. Фильтр по модельному дню вдобавок пропускал бы битые строки — у них дата не извлекается.

Две оси времени тут не совпадают намеренно. Ось модельного времени начинается в D0 и к реальному календарю не привязана; пакетный режим проигрывает две недели модельного мира за минуты реальных. Поэтому у переобработки и у гигиены диска разные часы, и обслуживают их разные средства. Декларативный TTL по модельной дате был бы просто сломан: он отсчитывает срок от реального «сейчас» и удалял бы эталонные дни прямо на входе.

В бою слой сырья иногда собирают на движке Null — тогда он не хранится вовсе. Такой вариант отвергнут: на стенде сырьё нужно для отладки, поэтому окно, а не ноль.

Таблица ошибок

ods.event_errors держит строки, не прошедшие строгий приём, вместе с их сырым текстом и метаданными доставки. Ключи её собственные, потому что у брака нет разобранных полей: шардируется cityHash64 сырой строки — ClientID у строки, которая не разобралась, взять неоткуда; нарезается по дню загрузки, как и сырьё; живёт месяц. Дольше сырья — намеренно: если брак истекает вместе с ним, разбираться к моменту разбирательства будет уже нечем.

Раскладка DDL

Файлы лежат в sql/ddl/ и применяются по порядку имён. Один файл — это слой и роль: статичные объекты отдельно от матвью.

Файл	Что в нём
00-databases.sql	базы слоёв
10-stg-tables.sql	Kafka-таблица, локальная и распределённая таблицы сырья
11-stg-views.sql	матвью, наполняющая сырьё из Kafka
20-ods-tables.sql	типизированное событие и таблица ошибок
21-ods-views.sql	матвью разбора: сырьё в событие и в ошибки

Матвью принадлежит слою своей цели, а не источника: разбор из STG в ODS лежит среди файлов ODS, потому что наполняет ODS.

Применение — двумя одноразовыми сервисами при make up, по образцу уже работающих airflow-init и superset-init. Сначала kafka-init создаёт топик hits с двумя партициями, затем clickhouse-init дожидается его завершения и применяет файлы с ноды 1, ON CLUSTER. Порядок страхует от автосоздания топика брокером с одной партицией: у потребителя librdkafka разрешение на автосоздание по умолчанию выключено, так что случиться это не обязано, но урок «обе ноды читают топик» умирает тихо, и полагаться на умолчание клиента здесь не стоит.

Повторный make up поверх живого тома проходит зелёным: весь DDL идёт через CREATE ... IF NOT EXISTS. Оборотная сторона — изменённый объект тем же запуском не применяется, причём молча. Отдельного механизма для этого нет и не нужно: правка существующего DDL случается, только пока стенд пишут, а лекарство уже есть — make clean && make up. Мир регенерируется, сырьё живёт трое суток, терять нечего.

Карта таблиц

Ниже — то, что закладывает этап 2. В репозитории этих объектов пока нет.

Слой	Объект	Что это
STG	stg.hits_raw_kafka	чтец топика hits, формат RawBLOB
STG	stg.hits_raw_rep / _dist	сырая строка сообщения плюс метаданные доставки
STG	stg.hits_raw_mv	наполняет сырьё из чтеца
ODS	ods.event_rep / _dist	типизированное широкое событие
ODS	ods.event_errors_rep / _dist	строки, не прошедшие строгий приём
ODS	ods.event_mv, ods.event_errors_mv	разбор сырья в событие и в ошибки

Слои DDS и DM появляются на следующих этапах; их состав задан разделом 7 мастер-спеки и переносится сюда по мере постройки.