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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Посты и Голоса

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

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

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

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

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

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

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 объектов на одну строку. Эту связь лучше оставить как отдельные таблицы или с добавленными статистиками.

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

Посты и 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, соединяя нашу статистику дубликатов с нашими постами.

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

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

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

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

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

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

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

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

Орchestrating и планирование денормализации

Пакетный процесс

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

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

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

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

Стриминг

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