メインコンテンツまでスキップ
メインコンテンツまでスキップ

マテリアライズドビューを使用して高速な時系列分析のためのロールアップを構築する

このチュートリアルでは、マテリアライズドビューを使用して、高ボリュームのイベントテーブルから事前集計されたロールアップを維持する方法を示します。 3つのオブジェクトを作成します。生テーブル、ロールアップテーブル、およびロールアップに自動的に書き込むマテリアライズドビューです。

このパターンを使用するタイミング​

次の場合にこのパターンを使用します：

  • 追加専用のイベントストリーム（クリック、ページビュー、IoT、ログ）がある。
  • ほとんどのクエリは、時間範囲の集計（分/時間/日単位）である。
  • すべての生行を再スキャンせずに、一貫したサブ秒の読み取りが必要。

生イベントテーブルの作成​

CREATE TABLE events_raw
(
    event_time   DateTime,
    user_id      UInt64,
    country      LowCardinality(String),
    event_type   LowCardinality(String),
    value        Float64
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_time, user_id)
TTL event_time + INTERVAL 90 DAY DELETE

ノート

  • PARTITION BY toYYYYMM(event_time)は、パーティションを小さく保ち、削除しやすくします。
  • ORDER BY (event_time, user_id)は、時間制限付きクエリと二次フィルタをサポートします。
  • LowCardinality(String)は、カテゴリカル次元のメモリを節約します。
  • TTLは、90日後に生データをクリーンアップします（保持要件に合わせて調整）。

ロールアップ（集計）テーブルの設計​

時間粒度に事前集計します。 最も一般的な分析ウィンドウに一致する粒度を選択してください。

CREATE TABLE events_rollup_1h
(
    bucket_start  DateTime,            -- start of the hour
    country       LowCardinality(String),
    event_type    LowCardinality(String),
    users_uniq    AggregateFunction(uniqExact, UInt64),
    value_sum     AggregateFunction(sum, Float64),
    value_avg     AggregateFunction(avg, Float64),
    events_count  AggregateFunction(count)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(bucket_start)
ORDER BY (bucket_start, country, event_type)

集計状態（例:AggregateFunction(sum, ...)）を保存します。これは部分集計をコンパクトに表現し、後でマージまたは確定できます。

ロールアップをポピュレートするマテリアライズドビューの作成​

このマテリアライズドビューは、events_rawへの挿入時に自動的に発火し、ロールアップに集計状態を書き込みます。

CREATE MATERIALIZED VIEW mv_events_rollup_1h
TO events_rollup_1h
AS
SELECT
    toStartOfHour(event_time) AS bucket_start,
    country,
    event_type,
    uniqExactState(user_id)   AS users_uniq,
    sumState(value)           AS value_sum,
    avgState(value)           AS value_avg,
    countState()              AS events_count
FROM events_raw
GROUP BY bucket_start, country, event_type;

サンプルデータの挿入​

いくつかのサンプルデータを挿入します：

INSERT INTO events_raw VALUES
    (now() - INTERVAL 4 SECOND, 101, 'US', 'view', 1),
    (now() - INTERVAL 3 SECOND, 101, 'US', 'click', 1),
    (now() - INTERVAL 2 SECOND, 202, 'DE', 'view', 1),
    (now() - INTERVAL 1 SECOND, 101, 'US', 'view', 1);

ロールアップのクエリ​

状態を読取り時にマージするか、確定することができます：

SELECT
    bucket_start,
    country,
    event_type,
    uniqExactMerge(users_uniq) AS users,
    sumMerge(value_sum)        AS value_sum,
    avgMerge(value_avg)        AS value_avg,
    countMerge(events_count)   AS events
FROM events_rollup_1h
WHERE bucket_start >= now() - INTERVAL 1 DAY
GROUP BY ALL
ORDER BY bucket_start, country, event_type;

ヒント

読み取りが常にロールアップにヒットすることを期待する場合は、1時間粒度で「プレーン」MergeTreeテーブルに確定された数値を書き込む2番目のマテリアライズドビューを作成できます。 状態はより柔軟性を提供し、一方で確定された数値はわずかに単純な読み取りを提供します。

パフォーマンス向上のために主キーのフィールドでフィルタリング​

EXPLAINコマンドを使用して、インデックスがデータをプルーニングするためにどのように使用されているかを確認できます：

EXPLAIN indexes=1
SELECT *
FROM events_rollup_1h
WHERE bucket_start BETWEEN now() - INTERVAL 3 DAY AND now()
  AND country = 'US';

    ┌─explain────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
1.  │ Expression ((Project names + Projection))                                                                                          │
2.  │   Expression                                                                                                                       │
3.  │     ReadFromMergeTree (default.events_rollup_1h)                                                                                   │
4.  │     Indexes:                                                                                                                       │
5.  │       MinMax                                                                                                                       │
6.  │         Keys:                                                                                                                      │
7.  │           bucket_start                                                                                                             │
8.  │         Condition: and((bucket_start in (-Inf, 1758550242]), (bucket_start in [1758291042, +Inf)))                                 │
9.  │         Parts: 1/1                                                                                                                 │
10. │         Granules: 1/1                                                                                                              │
11. │       Partition                                                                                                                    │
12. │         Keys:                                                                                                                      │
13. │           toYYYYMM(bucket_start)                                                                                                   │
14. │         Condition: and((toYYYYMM(bucket_start) in (-Inf, 202509]), (toYYYYMM(bucket_start) in [202509, +Inf)))                     │
15. │         Parts: 1/1                                                                                                                 │
16. │         Granules: 1/1                                                                                                              │
17. │       PrimaryKey                                                                                                                   │
18. │         Keys:                                                                                                                      │
19. │           bucket_start                                                                                                             │
20. │           country                                                                                                                  │
21. │         Condition: and((country in ['US', 'US']), and((bucket_start in (-Inf, 1758550242]), (bucket_start in [1758291042, +Inf)))) │
22. │         Parts: 1/1                                                                                                                 │
23. │         Granules: 1/1                                                                                                              │
    └────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

上記のクエリ実行計画では、以下の3種類のインデックスが使用されていることが示されています： MinMaxインデックス、パーティションインデックス、および主キーインデックスです。 各インデックスは、主キーに指定されたフィールド(bucket_start, country, event_type)を利用しています。 最良のフィルタリング性能を得るためには、クエリが主キーのフィールドを使用してデータをプルーニングしていることを確認してください。

一般的なバリエーション​

  • 異なる粒度：日次ロールアップを追加：
CREATE TABLE events_rollup_1d
(
    bucket_start Date,
    country      LowCardinality(String),
    event_type   LowCardinality(String),
    users_uniq   AggregateFunction(uniqExact, UInt64),
    value_sum    AggregateFunction(sum, Float64),
    value_avg    AggregateFunction(avg, Float64),
    events_count AggregateFunction(count)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(bucket_start)
ORDER BY (bucket_start, country, event_type);

次に、2番目のマテリアライズドビュー：

CREATE MATERIALIZED VIEW mv_events_rollup_1d
TO events_rollup_1d
AS
SELECT
    toDate(event_time) AS bucket_start,
    country,
    event_type,
    uniqExactState(user_id),
    sumState(value),
    avgState(value),
    countState()
FROM events_raw
GROUP BY ALL;
  • 圧縮：生テーブルの大きなカラムにコーデックを適用（例：Codec(ZSTD(3))）。
  • コスト管理：重い保持を生テーブルにプッシュし、長期間のロールアップを保持。
  • バックフィル：履歴データをロードする際は、events_rawに挿入し、マテリアライズドビューが自動的にロールアップを構築させます。既存の行については、適切であればマテリアライズドビューの作成時にPOPULATEを使用するか、INSERT SELECTを使用します。

クリーンアップと保持​

  • 生のTTLを増やす（例：30/90日）が、ロールアップをより長く保持する（例：1年）。
  • TTLを使用して古いパーツを安価なストレージに移動することもできます（階層化が有効な場合）。

トラブルシューティング​

  • マテリアライズドビューが更新されていない？挿入がevents_raw（ロールアップテーブルではなく）に行われていること、マテリアライズドビューのターゲットが正しいことを確認してください（TO events_rollup_1h）。
  • クエリが遅い？ロールアップにヒットしていることを確認し（ロールアップテーブルを直接クエリ）、時間フィルタがロールアップ粒度に一致していることを確認してください。
  • バックフィルの不一致？SYSTEM FLUSH LOGSを使用し、挿入とマージを確認するためにsystem.query_log / system.partsをチェックしてください。