ClickHouse Cloud: Join テーブルエンジンによる高速で更新可能なルックアップ

ClickHouse における辞書 #

トランザクション系やイベントベースのデータソースから ClickHouse のような分析用データベースにデータを移行する際、多くの場合は Kimball 手法 に沿ったディメンショナルスキーマの設計を検討することになります。

ディメンショナルモデリング では、常にファクト (メジャー) とディメンション (コンテキスト) の概念が使用されます。ファクトは通常 (例外もありますが) 集計可能な数値データであり、ディメンションはファクトを定義する階層と記述子のグループです。

したがってファクトテーブルは基本的にイミュータブルであり、データは追記されます。一方、ディメンションテーブルはサイズが小さく、(頻度は低いものの) 更新の対象となります (緩やかに変化するディメンション (SCD: slowly changing dimensions))。分析クエリを実行する際には、これらのディメンションをファクトテーブルに対して結合する必要があります。

ClickHouse でこれを行う一般的な方法の 1 つは、ディメンションデータのコピーを 辞書 (Dictionary) としてメモリ上に保持することです。このアプローチにより Direct Join が可能になり、結合パフォーマンスの最適化策として推奨 されています。

辞書を設定するには、とりわけ SOURCE 属性と LIFETIME 属性を指定します。ClickHouse はソースから最新データを取得し、LIFETIME に基づいて辞書を更新する頻度を決定します。しかし、一部のユーザーから「辞書を通常のテーブルのように更新する方法はないのか」という質問を受けました。実のところ、別の特殊なテーブルエンジンを使用することでこれを実現する方法が存在します。

Join テーブルエンジン #

まさにここで必要となるのが Join テーブルエンジン です。これはテーブル定義で指定された特定の結合タイプに合わせてデータを配置するインメモリ構造であり、永続化レイヤーによってバックアップされています。Join テーブルを設定する際は、以下を構成する必要があります。

• 結合の厳密性 (join strictness)
• 結合タイプ (join type)
• 結合で使用するキーカラム

結合の厳密性 (Join strictness) #

指定できる値は ANY または ALL です。ALL の場合、一致するすべての行が Join テーブルから取得されます。ANY の場合は最新の 1 行のみが取得されます。

つまり、ANY タイプを使用すると、キー単位で INSERT が実質的な UPSERT になります。同じキーを持つ別の行を挿入することで、ディメンションの行を更新できるのです。

結合タイプ (Join Type) #

INNER、LEFT、RIGHT など、ClickHouse の結合タイプのいずれかを指定します。ディメンショナルモデリングでは、ほとんどの場合 LEFT を使用します。

Join テーブルへのクエリ #

Join テーブルは、通常のテーブルと同様に SELECT を使ってクエリできるだけでなく、さらに 2 つの活用方法があります。

1. テーブル定義と一致する結合パラメータを持つ JOIN クエリ内に Join テーブルを配置すると、ClickHouse は自動的に Direct Join アルゴリズムを採用します。
2. joinGet を使用して、指定したキーに対応する値をルックアップできます。これは辞書の dictGet とまったく同様に機能します。

オープンソース版実装における課題 #

では、ANY LEFT 結合条件を持つ Join テーブルを使えば、タイプ 1 の緩やかに変化するディメンション (SCD Type 1: 上書き) を実装するのにうってつけではないでしょうか。指定したキーの値を更新でき、高速な結合も実現できます。それでは、なぜ常にこれが使われないのでしょうか。

実は、オープンソース版 ClickHouse における Join テーブルエンジンの実装には、このユースケースでの採用を難しくするいくつかのデメリットがあります。

1. Join テーブルが分散配置されないこと: 各クラスターノードがテーブルのコピー/バージョンを独自に保持する必要があります。
2. 永続化レイヤーが頻繁な挿入/更新を想定して構築されていないこと: Join テーブルエンジンは、ディスク上のテーブルデータディレクトリ内に圧縮された Native 形式の .bin ファイルとしてデータを永続化します (INSERT バッチごとに 1 ファイル)。サーバー起動時にはこれらのファイルがシーケンシャルに読み戻され、インメモリの HashJoin ハッシュテーブルが再構築されます。つまり、更新のたびに番号付きの新しい .bin ファイルが作成されます。バックグラウンドでのコンパクション処理が存在しないため、ファイルが自動的にマージされることはありません。時間の経過とともに、これはパフォーマンスの低下を招きます。

ClickHouse Cloud での実装 #

ClickHouse Cloud では、これらの問題が非常にエレガントに解決されています。 ClickHouse Cloud における Join テーブルは、実際には MergeTree ファミリーのバッキングテーブルを備えた SharedJoin テーブルとして透過的に実装されています。

• ALL 結合の場合、MergeTree テーブルになります。
• ANY 結合の場合、ReplacingMergeTree テーブルになります。

これらのテーブルは system.tables で確認できます。基盤となるテーブルの命名規則は .inner_id.SharedJoin.<Join テーブルの UUID> です。

注意: join_any_take_last_row という設定項目がありますが、これは適用されません。

インメモリテーブルは、Join テーブルへの挿入時 (最新データのみを選択するフィルター付き) および起動時のテーブル読み込み時に、永続化された (基盤となる) テーブルからのクエリ (ANY 結合の場合は FINAL を含む) によってデータが投入されます。

例: データエンリッチメント #

最も実用的なユースケースは、おそらく ANY LEFT 結合を使用したエンリッチメントやディメンショナルモデリングです。この具体的なケースを説明するために、ClickHouse のドキュメントにある例を少し変更して使ってみましょう。

1-- Create the fact table and insert some data
2CREATE OR REPLACE TABLE id_val (
3    `id` UInt32,
4    `val` UInt32
5) ENGINE = MergeTree
6ORDER BY (id);
7
8INSERT INTO id_val VALUES
9    (1, 11), (2, 12), (3, 13);
10
11-- Creating the right-side Join table:
12CREATE OR REPLACE TABLE id_val_join (
13    `id` UInt32,
14    `val` UInt8
15) ENGINE = Join(ANY, LEFT, id);
16
17-- Insert some values
18INSERT INTO id_val_join VALUES
19    (1, 21), (1, 22), (3, 23);
20
21-- Enrichment query
22SELECT *
23FROM id_val
24ANY LEFT JOIN id_val_join USING (id);

1┌─id─┬─val─┬─id_val_join.val─┐
21. │  1 │  11 │              22 │
32. │  2 │  12 │               0 │
43. │  3 │  13 │              23 │
5   └────┴─────┴─────────────────┘

ここで、キー 1 のエントリを upsert したときに、Join テーブルとその基盤となるテーブルで何が起きるかを確認してみましょう。

1-- And another insert
2INSERT INTO id_val_join VALUES (1,42);

基盤となるテーブルを調べます。

1SELECT database, name, uuid, engine
2FROM system.tables
3WHERE name = 'id_val_join'
4FORMAT Vertical;

1database:                         default
2name:                             id_val_join
3uuid:                             64f169ee-977d-46c2-b067-580fdf8c1d4b
4engine:                           SharedJoin

Join テーブル側では重複が排除され、最新のエントリのみが保持されます。

1SELECT * FROM id_val_join;

1┌─id─┬─val─┐
21. │  3 │  23 │
32. │  1 │  42 │
4   └────┴─────┘

UUID から基盤となる ReplacingMergeTree テーブルを特定して確認すると、マージによって解消されるまで重複が残っていることがわかります。

1SELECT * FROM default.`.inner_id.SharedJoin.64f169ee-977d-46c2-b067-580fdf8c1d4b`;

1┌─id─┬─val─┐
21. │  1 │  22 │
32. │  3 │  23 │
43. │  1 │  42 │
5   └────┴─────┘

最後にエンリッチメントクエリを再実行すると、更新されたディメンションのエントリが結果に反映されていることが確認できます。

1SELECT *
2FROM id_val
3ANY LEFT JOIN id_val_join USING (id);

1┌─id─┬─val─┬─id_val_join.val─┐
21. │  1 │  11 │              42 │
32. │  2 │  12 │               0 │
43. │  3 │  13 │              23 │
5   └────┴─────┴─────────────────┘

新しい行や行セットを挿入するたびに、以下の処理が行われます。

• データは基盤となる ReplacingMergeTree テーブルに挿入されます。
• インメモリのデータ表現は、Join テーブルへの挿入時 (最新データのみを選択するためのブロック ID フィルター付き) および起動時のテーブル読み込み時に更新されます。
• このクエリには FINAL も適用されるため、インメモリの Join テーブルに重複が含まれることは決してありません。
• join_any_take_last_row は無視されます。常に最新のエントリが取得されます。

まとめ #

• ClickHouse の Join テーブルエンジンは、事前計算されたハッシュマップを提供し、JOIN の高速化に役立ちます。
• 辞書と同様に、Join テーブルはメモリ上に保持されます。ただし、永続化レイヤー (ファイルへの保存) によってバックアップされています。
• ClickHouse Cloud では、Join テーブルが自動的にクラスター化され、完全な MergeTree テーブルによって裏付けられるため、頻繁な更新にも適しています。
• 特に ClickHouse Cloud でディメンショナルモデリングを行う際は、Join(ANY, LEFT, id) をご活用ください。upsert、重複排除、データのコンパクションは、基盤となる ReplacingMergeTree によってすべて自動的に処理されます。