Денормализация данных

Денормализация данных — это подход в ClickHouse, при котором используются «плоские» таблицы для минимизации задержки выполнения запросов за счёт избегания соединений (JOIN).

Сравнение нормализованных и денормализованных схем

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

Этот процесс снижает необходимость сложных JOIN во время выполнения запроса и может существенно ускорить операции чтения, что делает его идеальным для приложений с высокой нагрузкой на чтение и сложными запросами. Однако он может увеличить сложность операций записи и обслуживания, поскольку любые изменения в дублированных данных должны быть распространены по всем их копиям для поддержания согласованности.

Распространённый приём, получивший популярность благодаря NoSQL-решениям, заключается в денормализации данных при отсутствии поддержки JOIN, фактически сохраняя всю статистику или связанные строки в родительской строке в виде столбцов и вложенных объектов. Например, в примерной схеме данных для блога мы можем хранить все Comments как Array объектов в соответствующих постах.

Когда использовать денормализацию

В целом мы рекомендуем использовать денормализацию в следующих случаях:

• Денормализуйте таблицы, которые изменяются редко, или для которых допустима задержка перед доступностью данных для аналитических запросов, то есть данные могут быть полностью перезагружены пакетно.
• Избегайте денормализации связей «многие ко многим». Это может привести к необходимости обновлять большое число строк при изменении одной исходной строки.
• Избегайте денормализации связей с высокой кардинальностью. Если каждая строка в таблице имеет тысячи связанных записей в другой таблице, их необходимо будет представить в виде Array — либо примитивного типа, либо кортежей. Как правило, использование массивов с более чем 1000 кортежей не рекомендуется.
• Вместо денормализации всех столбцов в виде вложенных объектов рассмотрите вариант денормализовать только статистику, используя материализованные представления (см. ниже).

Нет необходимости денормализовывать всю информацию — только ключевые данные, к которым нужен частый доступ.

Работу по денормализации можно выполнять либо в ClickHouse, либо в upstream‑системах, например с использованием Apache Flink.

Избегайте денормализации для часто обновляемых данных

В ClickHouse денормализация — это один из вариантов оптимизации производительности запросов, но применять её следует осторожно. Если данные часто меняются и их нужно обновлять почти в режиме реального времени, от этого подхода лучше отказаться. Используйте его, если основная таблица в основном только пополняется (append-only) или может периодически полностью перезагружаться пакетами, например ежедневно.

У этого подхода есть одна ключевая проблема — производительность записи и обновления данных. Более конкретно, денормализация по сути переносит выполнение join-ов с момента выполнения запроса на момент ингестии. Хотя это может значительно повысить производительность запросов, это усложняет ингестию и означает, что конвейеры данных должны повторно вставлять строку в ClickHouse, если изменилась любая из строк, использованных для её формирования. Это может означать, что изменение одной исходной строки потенциально приводит к необходимости обновления множества строк в ClickHouse. В сложных схемах, где строки формируются из сложных соединений, изменение одной строки во вложенном компоненте join-а может потенциально привести к необходимости обновления миллионов строк.

Добиться этого в реальном времени часто нереалистично и требует значительных инженерных усилий по двум причинам:

1. Инициация корректных операторов join при изменении строки таблицы. В идеале это не должно приводить к обновлению всех объектов, участвующих в join-е, а только тех, которые были затронуты. Эффективная модификация join-ов так, чтобы они фильтровали только нужные строки и сохраняли производительность при высокой пропускной способности, требует внешних инструментов или дополнительных доработок.
2. Обновления строк в ClickHouse необходимо тщательно контролировать, что добавляет дополнительную сложность.

Поэтому чаще используется пакетный процесс обновления, при котором все денормализованные объекты периодически перезагружаются.

Практические примеры денормализации

Рассмотрим несколько практических примеров, когда денормализация может иметь смысл, а также случаи, когда предпочтительнее использовать альтернативные подходы.

Предположим, у нас есть таблица Posts, которая уже денормализована и содержит статистику, такую как AnswerCount и CommentCount — исходные данные предоставляются в таком виде. На практике мы можем захотеть нормализовать эту информацию, так как она, вероятно, будет часто меняться. Многие из этих столбцов также доступны через другие таблицы, например, комментарии к посту доступны через столбец PostId и таблицу Comments. В рамках данного примера мы предполагаем, что посты перезагружаются пакетным процессом.

Мы также рассматриваем только денормализацию других таблиц в Posts, так как считаем её нашей основной таблицей для аналитики. Денормализация в обратном направлении также будет уместна для некоторых запросов, с учётом тех же соображений, описанных выше.

Для каждого из следующих примеров предположим, что существует запрос, которому требуется использовать обе таблицы в соединении.

Posts и Votes

Голосования по постам представлены отдельными таблицами. Оптимизированная схема для этого показана ниже, а также команда INSERT для загрузки данных:

CREATE TABLE votes
(
        `Id` UInt32,
        `PostId` Int32,
        `VoteTypeId` UInt8,
        `CreationDate` DateTime64(3, 'UTC'),
        `UserId` Int32,
        `BountyAmount` UInt8
)
ENGINE = MergeTree
ORDER BY (VoteTypeId, CreationDate, PostId)

INSERT INTO votes SELECT * FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/votes/*.parquet')

0 rows in set. Elapsed: 26.272 sec. Processed 238.98 million rows, 2.13 GB (9.10 million rows/s., 80.97 MB/s.)

На первый взгляд, эти поля можно рассматривать как кандидатов для денормализации в таблице posts. У такого подхода есть несколько сложностей.

Голоса к постам добавляются часто. Хотя со временем их количество в пересчёте на один пост может снижаться, следующий запрос показывает, что у нас около 40 тысяч голосов в час по 30 тысячам постов.

SELECT round(avg(c)) AS avg_votes_per_hr, round(avg(posts)) AS avg_posts_per_hr
FROM
(
        SELECT
        toStartOfHour(CreationDate) AS hr,
        count() AS c,
        uniq(PostId) AS posts
        FROM votes
        GROUP BY hr
)

┌─avg_votes_per_hr─┬─avg_posts_per_hr─┐
│               41759 │         33322 │
└──────────────────┴──────────────────┘

Эту проблему можно решить с помощью пакетной обработки, если допустима задержка, но при этом нам всё равно нужно обрабатывать обновления, если только мы не будем периодически перезагружать все посты (что вряд ли желательно).

Более серьёзная проблема в том, что у некоторых постов чрезвычайно большое количество голосов:

SELECT PostId, concat('https://stackoverflow.com/questions/', PostId) AS url, count() AS c
FROM votes
GROUP BY PostId
ORDER BY c DESC
LIMIT 5

┌───PostId─┬─url──────────────────────────────────────────┬─────c─┐
│ 11227902 │ https://stackoverflow.com/questions/11227902 │ 35123 │
│   927386 │ https://stackoverflow.com/questions/927386   │ 29090 │
│ 11227809 │ https://stackoverflow.com/questions/11227809 │ 27475 │
│   927358 │ https://stackoverflow.com/questions/927358   │ 26409 │
│  2003515 │ https://stackoverflow.com/questions/2003515  │ 25899 │
└──────────┴──────────────────────────────────────────────┴───────┘

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

Пользователи и бейджи

Теперь рассмотрим наши Users и Badges:

Сначала вставим данные следующей командой:

CREATE TABLE users
(
    `Id` Int32,
    `Reputation` LowCardinality(String),
    `CreationDate` DateTime64(3, 'UTC') CODEC(Delta(8), ZSTD(1)),
    `DisplayName` String,
    `LastAccessDate` DateTime64(3, 'UTC'),
    `AboutMe` String,
    `Views` UInt32,
    `UpVotes` UInt32,
    `DownVotes` UInt32,
    `WebsiteUrl` String,
    `Location` LowCardinality(String),
    `AccountId` Int32
)
ENGINE = MergeTree
ORDER BY (Id, CreationDate)

CREATE TABLE badges
(
    `Id` UInt32,
    `UserId` Int32,
    `Name` LowCardinality(String),
    `Date` DateTime64(3, 'UTC'),
    `Class` Enum8('Gold' = 1, 'Silver' = 2, 'Bronze' = 3),
    `TagBased` Bool
)
ENGINE = MergeTree
ORDER BY UserId

INSERT INTO users SELECT * FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/users.parquet')

0 строк в наборе. Прошло: 26.229 сек. Обработано 22.48 млн строк, 1.36 ГБ (857.21 тыс. строк/с., 51.99 МБ/с.)

INSERT INTO badges SELECT * FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/badges.parquet')

0 строк в наборе. Прошло: 18.126 сек. Обработано 51.29 млн строк, 797.05 МБ (2.83 млн строк/с., 43.97 МБ/с.)

Хотя пользователи могут часто получать бейджи, маловероятно, что этот набор данных нам нужно будет обновлять чаще одного раза в день. Связь между бейджами и пользователями — «один ко многим». Может быть, мы просто денормализуем бейджи на пользователей как список кортежей? Хотя это возможно, быстрая проверка максимального количества бейджей на одного пользователя показывает, что это не лучший вариант:

SELECT UserId, count() AS c FROM badges GROUP BY UserId ORDER BY c DESC LIMIT 5

┌─UserId─┬─────c─┐
│  22656 │ 19334 │
│   6309 │ 10516 │
│ 100297 │  7848 │
│ 157882 │  7574 │
│  29407 │  6512 │
└────────┴───────┘

Вряд ли реалистично денормализовать 19 000 объектов в одну строку. Эту связь, вероятно, лучше оставить в виде отдельных таблиц или добавить к ней статистику.

Мы можем захотеть денормализовать статистику из badges в users, например количество значков. Мы рассмотрим такой пример при использовании словарей для этого набора данных на этапе вставки.

Posts и PostLinks

PostLinks соединяют Posts, которые пользователи считают связанными или дубликатами. Следующий запрос показывает схему и команду загрузки:

CREATE TABLE postlinks
(
  `Id` UInt64,
  `CreationDate` DateTime64(3, 'UTC'),
  `PostId` Int32,
  `RelatedPostId` Int32,
  `LinkTypeId` Enum('Linked' = 1, 'Duplicate' = 3)
)
ENGINE = MergeTree
ORDER BY (PostId, RelatedPostId)

INSERT INTO postlinks SELECT * FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/stackoverflow/parquet/postlinks.parquet')

0 rows in set. Elapsed: 4.726 sec. Processed 6.55 million rows, 129.70 MB (1.39 million rows/s., 27.44 MB/s.)

Можем подтвердить, что ни одна запись не содержит чрезмерного количества ссылок, которое бы препятствовало денормализации:

SELECT PostId, count() AS c
FROM postlinks
GROUP BY PostId
ORDER BY c DESC LIMIT 5

┌───PostId─┬───c─┐
│ 22937618 │ 125 │
│  9549780 │ 120 │
│  3737139 │ 109 │
│ 18050071 │ 103 │
│ 25889234 │  82 │
└──────────┴─────┘

Аналогично, такие связи не относятся к часто происходящим событиям:

SELECT
  round(avg(c)) AS avg_votes_per_hr,
  round(avg(posts)) AS avg_posts_per_hr
FROM
(
  SELECT
  toStartOfHour(CreationDate) AS hr,
  count() AS c,
  uniq(PostId) AS posts
  FROM postlinks
  GROUP BY hr
)

┌─avg_votes_per_hr─┬─avg_posts_per_hr─┐
│                54 │                    44     │
└──────────────────┴──────────────────┘

Мы будем использовать это в качестве примера денормализации ниже.

Простой статистический пример

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

CREATE TABLE posts_with_duplicate_count
(
  `Id` Int32 CODEC(Delta(4), ZSTD(1)),
   ... -- другие столбцы
   `DuplicatePosts` UInt16
) ENGINE = MergeTree
ORDER BY (PostTypeId, toDate(CreationDate), CommentCount)

Чтобы заполнить эту таблицу, мы выполняем запрос INSERT INTO SELECT, объединяя таблицу со статистикой по дубликатам с таблицей постов.

INSERT INTO posts_with_duplicate_count SELECT
    posts.*,
    DuplicatePosts
FROM posts AS posts
LEFT JOIN
(
    SELECT PostId, countIf(LinkTypeId = 'Duplicate') AS DuplicatePosts
    FROM postlinks
    GROUP BY PostId
) AS postlinks ON posts.Id = postlinks.PostId

Использование сложных типов для отношений «один-ко-многим»

Для выполнения денормализации нам часто необходимо использовать сложные типы. Если денормализуется отношение «один-к-одному» с небольшим количеством столбцов, пользователи могут просто добавить их как столбцы с их исходными типами, как показано выше. Однако для более крупных объектов это часто нежелательно, а для отношений «один-ко-многим» — и вовсе невозможно.

В случаях сложных объектов или отношений «один-ко-многим» пользователи могут использовать:

• именованные Tuple — позволяют представить связанную структуру как набор столбцов;
• Array(Tuple) или Nested — массив именованных Tuple, также известный как Nested, где каждая запись представляет объект. Применимо к отношениям «один-ко-многим».

В качестве примера ниже мы показываем, как денормализовать PostLinks на Posts.

Каждый пост может содержать несколько ссылок на другие посты, как показано ранее в схеме PostLinks. В виде типа Nested мы можем представить эти связанные и дублирующиеся посты следующим образом:

SET flatten_nested=0
CREATE TABLE posts_with_links
(
  `Id` Int32 CODEC(Delta(4), ZSTD(1)),
   ... -остальные столбцы
   `LinkedPosts` Nested(CreationDate DateTime64(3, 'UTC'), PostId Int32),
   `DuplicatePosts` Nested(CreationDate DateTime64(3, 'UTC'), PostId Int32),
) ENGINE = MergeTree
ORDER BY (PostTypeId, toDate(CreationDate), CommentCount)

Обратите внимание на использование настройки flatten_nested=0. Мы рекомендуем отключить развёртывание вложенных данных.

Мы можем выполнить эту денормализацию с помощью запроса INSERT INTO SELECT с OUTER JOIN:

INSERT INTO posts_with_links
SELECT
    posts.*,
    arrayMap(p -> (p.1, p.2), arrayFilter(p -> p.3 = 'Linked' AND p.2 != 0, Related)) AS LinkedPosts,
    arrayMap(p -> (p.1, p.2), arrayFilter(p -> p.3 = 'Duplicate' AND p.2 != 0, Related)) AS DuplicatePosts
FROM posts
LEFT JOIN (
    SELECT
         PostId,
         groupArray((CreationDate, RelatedPostId, LinkTypeId)) AS Related
    FROM postlinks
    GROUP BY PostId
) AS postlinks ON posts.Id = postlinks.PostId

0 rows in set. Elapsed: 155.372 sec. Processed 66.37 million rows, 76.33 GB (427.18 thousand rows/s., 491.25 MB/s.)
Peak memory usage: 6.98 GiB.

Обратите внимание на время выполнения. Нам удалось денормализовать 66 млн строк примерно за 2 минуты. Как мы увидим позже, эту операцию можно запускать по расписанию.

Обратите внимание на использование функций groupArray для агрегирования PostLinks в массив для каждого PostId перед выполнением соединения. Этот массив затем фильтруется в два списка: LinkedPosts и DuplicatePosts, при этом из внешнего соединения также исключаются любые пустые результаты.

Мы можем выбрать несколько строк, чтобы увидеть нашу новую денормализованную структуру:

SELECT LinkedPosts, DuplicatePosts
FROM posts_with_links
WHERE (length(LinkedPosts) > 2) AND (length(DuplicatePosts) > 0)
LIMIT 1
FORMAT Vertical

Row 1:
──────
LinkedPosts:    [('2017-04-11 11:53:09.583',3404508),('2017-04-11 11:49:07.680',3922739),('2017-04-11 11:48:33.353',33058004)]
DuplicatePosts: [('2017-04-11 12:18:37.260',3922739),('2017-04-11 12:18:37.260',33058004)]

Оркестрация и планирование денормализации

Пакетная обработка

Эффективное использование денормализации требует процесса преобразования, в рамках которого её можно выполнять и оркестрировать.

Выше мы показали, как ClickHouse можно использовать для выполнения этого преобразования после загрузки данных с помощью INSERT INTO SELECT. Это подходит для периодических пакетных преобразований.

У пользователей есть несколько вариантов оркестрации этого процесса в ClickHouse при условии, что их устраивает периодическая пакетная загрузка данных:

• Обновляемые материализованные представления — обновляемые материализованные представления можно использовать для периодического запуска запроса с отправкой результатов в целевую таблицу. При выполнении запроса представление гарантирует атомарное обновление целевой таблицы. Это предоставляет встроенный в ClickHouse способ планирования этой работы.
• Внешние инструменты — использование таких инструментов, как dbt и Airflow, для периодического запуска преобразования. Интеграция ClickHouse с dbt гарантирует, что это выполняется атомарно: создаётся новая версия целевой таблицы, которая затем атомарно меняется местами с версией, обслуживающей запросы (через команду EXCHANGE).

Потоковая обработка

Пользователи могут также выполнять этот процесс вне ClickHouse, до вставки данных, используя потоковые технологии, такие как Apache Flink. Либо можно использовать инкрементальные материализованные представления для выполнения этого процесса по мере вставки данных.