データの非正規化

非正規化でクエリ性能を向上させる方法

データの非正規化は、JOINを回避してクエリのレイテンシを最小限に抑えるために、ClickHouseでフラットなテーブルを使用する手法です。

​

正規化スキーマと非正規化スキーマの比較

データの非正規化とは、特定のクエリパターンに対するデータベース性能を最適化するために、正規化のプロセスを意図的に逆転させることです。正規化されたデータベースでは、冗長性を最小限に抑え、データの整合性を確保するために、データは複数の関連テーブルに分割されます。非正規化では、テーブルを結合し、データを複製し、計算済みフィールドを単一のテーブルまたはより少ない数のテーブルに取り込むことで、冗長性を再び持たせます。つまり、JOIN をクエリ時ではなく挿入時に行う形へと実質的に移すことになります。

このプロセスにより、クエリ時の複雑な JOIN が不要になり、読み取り処理を大幅に高速化できるため、読み取り負荷が高く複雑なクエリを伴うアプリケーションに適しています。一方で、書き込み処理や保守の複雑さは増す可能性があります。整合性を維持するには、複製されたデータへの変更をすべてのインスタンスに反映させる必要があるためです。

NoSQL ソリューションによって広まった一般的な手法の 1 つに、JOIN をサポートしない環境でデータを非正規化し、すべての統計情報や関連する行を、カラムやネストされたオブジェクトとして親の行に格納する方法があります。たとえば、ブログのスキーマ例では、各投稿に属するすべての Comments を、オブジェクトの Array としてそれぞれの投稿に格納できます。

​

非正規化を使用するタイミング

一般に、次のような場合に非正規化を推奨します。

• 更新頻度が低いテーブル、または分析クエリで利用可能になるまでの遅延を許容できるテーブルを非正規化します。つまり、データをバッチで完全に再読み込みできる場合です。
• 多対多の関係を非正規化するのは避けてください。1 つの元データの行が変更されただけでも、多数の行を更新する必要が生じる可能性があります。
• カーディナリティの高い関係を非正規化するのは避けてください。あるテーブルの各行が、別のテーブルに数千件の関連エントリを持つ場合、それらは Array (プリミティブ型またはタプル) として表現する必要があります。一般に、1000 個を超えるタプルを含む配列は推奨されません。
• すべてのカラムをネストされたオブジェクトとして非正規化するのではなく、materialized view を使って統計情報のみを非正規化することも検討してください (以下を参照) 。

すべての情報を非正規化する必要はありません。頻繁にアクセスする必要がある重要な情報だけで十分です。

非正規化の処理は、ClickHouse 側でも上流側でも行えます。たとえば、Apache Flink を使用できます。

​

頻繁に更新されるデータでは非正規化を避ける

ClickHouse では、非正規化はクエリ性能を最適化するための選択肢の 1 つですが、慎重に使用する必要があります。データが頻繁に更新され、ほぼリアルタイムで反映する必要がある場合は、このアプローチは避けるべきです。これは、メインテーブルがほぼ追記専用である場合や、たとえば日次のように定期的にバッチで再読み込みできる場合に適しています。

このアプローチには、本質的に 1 つの大きな課題があります。書き込み性能とデータ更新です。より具体的には、非正規化では、データの JOIN の責任をクエリ時からインジェスト時へ実質的に移すことになります。これによりクエリ性能は大幅に向上する可能性がありますが、その一方でインジェストは複雑になり、それを構成している元の行のいずれかが変更された場合、データパイプラインがその行を ClickHouse に再挿入する必要が生じます。つまり、1 つの元データの行の変更によって、ClickHouse 内の多数の行を更新しなければならない可能性があります。複雑なスキーマでは、複雑な JOIN によって行が構成されているため、JOIN のネストされた部分内の 1 行の変更が、数百万行の更新を必要とする可能性もあります。

これをリアルタイムで実現するのは、多くの場合現実的ではなく、次の 2 つの課題があるため、相応のエンジニアリングが必要になります。

1. テーブルの行が変更されたときに、適切な JOIN 文をトリガーすること。理想的には、これによって JOIN 対象のすべてが更新されるのではなく、影響を受けたものだけが更新されるべきです。適切な行だけを効率的に絞り込めるように JOIN を調整し、さらに高スループット下でこれを実現するには、外部ツールまたは追加のエンジニアリングが必要です。
2. ClickHouse における行の更新は慎重に管理する必要があり、複雑さがさらに増します。

そのため、一般的には、非正規化されたデータを定期的にまとめて再読み込みするバッチ更新プロセスが採用されます。

​

非正規化の実践例

非正規化が適している実践的な例と、代わりに別のアプローチのほうが望ましい例をいくつか見ていきます。

Posts テーブルが、AnswerCount や CommentCount などの統計情報を含む形ですでに非正規化されているとします。元データはこの形式で提供されています。実際には、こうした情報は頻繁に更新される可能性が高いため、むしろ正規化したい場合もあります。これらのカラムの多くは他のテーブルからも取得できます。たとえば、投稿に対するコメントは PostId カラムと Comments テーブルを通じて参照できます。ここでは例として、投稿はバッチ処理で再読み込みされるものとします。

また、ここでは他のテーブルの内容を Posts に非正規化する場合だけを考えます。これは、分析における主テーブルを Posts とみなしているためです。逆方向に非正規化することも、一部のクエリでは適切です。その場合にも、上記と同じ考慮事項が当てはまります。

以下の各例では、両方のテーブルを JOIN で使用する必要があるクエリが存在すると仮定してください。

​

Posts and 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 テーブルで非正規化する候補に思えるかもしれません。ただし、このアプローチにはいくつかの課題があります。

posts には Votes が頻繁に追加されます。これは時間の経過とともに 1 post あたりでは減っていく可能性がありますが、次のクエリが示すように、30k posts に対して 1 時間あたり約 40k votes あります。

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 カラムはそのような統計の 1 つで、つまり賛成票の合計から反対票の合計を引いた値を表しています。理想的には、クエリ時に単純なルックアップでこうした統計を取得できるのが望ましいでしょう (辞書を参照) 。

​

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

ユーザーは頻繁にバッジを獲得する可能性がありますが、このデータセットを日次より高い頻度で更新する必要はおそらくありません。バッジとユーザーの関係は一対多です。では、バッジをタプルのリストとしてユーザー側に単純に非正規化すればよいのでしょうか。もちろん可能ではありますが、ユーザーごとのバッジ数の最大値をざっと確認すると、これは理想的ではないことがわかります:

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個のオブジェクトを1つの行に非正規化するのは、おそらく現実的ではありません。この関係は、別々のテーブルのままにしておくか、統計情報を追加するのがよいでしょう。

バッジに関する統計情報を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     │
└──────────────────┴──────────────────┘

以下では、これを非正規化の例として用います。

​

単純な統計の例

ほとんどの場合、非正規化では、親行に 1 つのカラムや統計値を追加するだけで済みます。たとえば、Posts に重複投稿数を追加したいだけであれば、必要なのはカラムを 1 つ追加することだけです。

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)

このテーブルにデータを投入するには、重複統計をpostsと結合する 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

​

一対多の関係で複合型を活用する

非正規化を行うには、多くの場合、複合型を活用する必要があります。一対一の関係を非正規化する場合、カラム数が少なければ、上で示したように、それらを元の型のままカラムとして追加するだけで済みます。ただし、オブジェクトが大きい場合はこの方法が適さないことが多く、一対多の関係では使用できません。

複雑なオブジェクトや一対多の関係では、次を使用できます。

• Named Tuples - 関連する構造をカラムの集合として表現できます。
• Array(Tuple) or Nested - 名前付きタプルの配列で、Nested とも呼ばれます。各エントリが 1 つのオブジェクトを表します。一対多の関係に適しています。

例として、以下では PostLinks を Posts に非正規化する方法を示します。

前述の PostLinks スキーマに示したように、各 post には他の post への複数のリンクを含めることができます。Nested 型では、これらのリンク先や重複先の post を次のように表現できます。

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 が使われている点に注意してください。ネストされたデータのフラット化は無効にすることを推奨します。

この非正規化は、OUTER JOIN クエリを使った INSERT INTO SELECT で実行できます:

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.

ここで処理時間に注目してください。約2分で6,600万行を非正規化できています。後ほど示すように、この処理はスケジュールできます。

JOIN の前に groupArray 関数を使って、PostLinks を各 PostId ごとの配列にまとめている点にも注目してください。次に、この配列を 2 つのサブリスト 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 でこれをオーケストレーションする方法はいくつかあります。

• リフレッシュ可能なマテリアライズドビュー - リフレッシュ可能なマテリアライズドビューを使用すると、結果をターゲットテーブルに書き出すクエリを定期的にスケジュールできます。クエリの実行時には、このビューによってターゲットテーブルがアトミックに更新されます。これにより、この処理をスケジュールするための ClickHouse ネイティブな手段が得られます。
• 外部ツール - dbt や Airflow などのツールを使って、変換を定期的にスケジュールすることもできます。dbt 向け ClickHouse 連携 を使うと、ターゲットテーブルの新しいバージョンを作成し、その後、クエリを受け付けているバージョンとアトミックに入れ替える (EXCHANGE コマンドを使用) ことで、これをアトミックに実行できます。

​

ストリーミング

あるいは、Apache Flink のようなストリーミング技術を使用して、ClickHouse の外部で、データを挿入する前にこの処理を実行することもできます。別の方法としては、増分 materialized view を使用して、データの挿入時にこの処理を実行することもできます。