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

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

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

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

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

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

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

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

Работу по денормализации можно выполнять как в ClickHouse, так и на стороне источника, например с помощью Apache Flink.

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

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

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

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

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

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

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

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

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

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

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', NOSIGN)

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. Однако у этого подхода есть несколько сложностей.

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

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 представляет собой такую статистику, то есть общее число положительных голосов минус число отрицательных. В идеале эту статистику можно было бы получать во время выполнения запроса с помощью простого поиска по ключу (см. словари).

Теперь рассмотрим наши 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', NOSIGN)

0 rows in set. Elapsed: 26.229 sec. Processed 22.48 million rows, 1.36 GB (857.21 thousand rows/s., 51.99 MB/s.)

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

0 rows in set. Elapsed: 18.126 sec. Processed 51.29 million rows, 797.05 MB (2.83 million rows/s., 43.97 MB/s.)

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

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 │
└────────┴───────┘

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

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

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', NOSIGN)

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)),
   ... -other columns
   `DuplicatePosts` UInt16
) ENGINE = MergeTree
ORDER BY (PostTypeId, toDate(CreationDate), CommentCount)

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

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)),
   ... -other columns
   `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.)
Пиковое потребление памяти: 6.98 GiB.

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

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

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

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, если допустим периодический процесс батч-загрузки:

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

В качестве альтернативы это можно выполнить вне ClickHouse, до вставки, с помощью технологий стриминга, таких как Apache Flink. Также для выполнения этого процесса по мере вставки данных можно использовать инкрементальные materialized views.