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

Как использовать денормализацию для повышения производительности запросов

Денормализация данных — это метод в 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.

​

Posts and Votes

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

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

Голоса к постам добавляются часто. Хотя со временем их число в расчёте на один пост может уменьшаться, следующий запрос показывает, что у нас около 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 and Badges

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

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, например число значков. Такой пример мы рассмотрим при использовании словарей для этого набора данных при вставке.

​

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)),
   ... -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.