TT Lab
はじめる
学ぶ 学習パス コース

ClickHouse — 列指向分析 DB を中身から

マテリアライズドビューが見たもの、見逃したもの

TT Labで続きを見る

目標

ソーステーブルにマテリアライズドビューをかけて、INSERTがターゲットテーブルに流れる様子を確認し、ビューが見えない3つのこと(ビューより前の行、マージ前の重複キー、JOINの右側のテーブルの変化)を、数値で再現します。

なぜ重要なのか

ClickHouseのマテリアライズドビューは「保存されたクエリ」ではなく、INSERTトリガーです。そのため、バックフィルを忘れると過去が空になり、2回バックフィルすると2倍になり、ターゲットテーブルを足さずに読むとマージのタイミングによって数値がぶれ、ディメンションテーブルのJOINは、遅れて来たディメンションを永久に取りこぼします。このラボの採点ツールは、書き込まれた数値をそのまま信用しません。ターゲットテーブルを、ソーステーブル(またはフィクスチャの生成式)から再計算した値と列ごとに突き合わせ、保存されたSELECTは読み取り専用でもう一度実行して、結果と読んだ行数を測ります。

ステップ

  1. データベースmvとテーブルmv.ordersを作成してください。列はorder_id UInt64, ts DateTime, shop LowCardinality(String), product_id UInt32, user_id UInt32, qty UInt32, price UInt32(この順序)、エンジンはMergeTree、ORDER BY (shop, ts)です。そして/opt/lab/fixtures/mv/orders_1.sql(9月1–15日、20万件)を1回だけ入れてください。
  2. ターゲットテーブルmv.daily_sales (day Date, shop LowCardinality(String), orders UInt64, revenue UInt64)をSummingMergeTree、ORDER BY (shop, day)で作成し、マテリアライズドビューmv.daily_sales_mvをTO mv.daily_salesで作成してください。toDate(ts) AS day, shop, count() AS orders, sum(qty * price) AS revenueをGROUP BY day, shopします。
  3. /opt/lab/fixtures/mv/orders_2.sql(9月16–30日、20万件)を入れた直後に、mv.ordersの行数をsource_orders、mv.daily_salesのsum(orders)をview_ordersとして書き込んでください(保存先: /root/ch/mv/gap.json)。
  4. ビューが見なかった1回目の分をmv.daily_salesにバックフィルして、ターゲットテーブルがソース全体を(shop, day)ごとに正確に持つようにしてください。2回目の分をもう一度入れてはいけません。
  5. mv.daily_salesからshop-3の日付ごとの注文数・売上を求めるクエリを書いてください(/root/ch/mv/q_daily.sql)。列はday, orders, revenueで、日付順です。マージ前でも正しくなるように、足して読む必要があります。
  6. ターゲットテーブルmv.shop_stats (shop LowCardinality(String), day Date, orders AggregateFunction(count), buyers AggregateFunction(uniqExact, UInt32))をAggregatingMergeTree、ORDER BY (shop, day)で、ビューmv.shop_stats_mvをTO mv.shop_statsで(countState()・uniqExactState(user_id))作成し、すでにあるソースを1回バックフィルしてください。そして、店舗ごとの1か月のユニーク購入者をmv.shop_statsから求めるクエリを書いてください(/root/ch/mv/q_buyers.sql)。列はshop, buyersで、店舗順です。
  7. mv.ordersと同じ列の末尾にstatus LowCardinality(String)を加えたmv.rawをENGINE = Nullで作成し、status = 'paid'の行だけをmv.ordersに送るビューmv.raw_mvを作成したあと、/opt/lab/fixtures/mv/raw.sqlを1回だけ入れてください。
  8. mv.products (product_id UInt32, category LowCardinality(String))(MergeTree、ORDER BY product_id)と、mv.cat_sales (category LowCardinality(String), orders UInt64)(SummingMergeTree、ORDER BY category)、そしてmv.ordersをmv.productsとINNER JOINしてカテゴリごとのcount()を送るビューmv.cat_sales_mvを作成してください。次に、products.sql→late.sql→products_new.sql(すべて/opt/lab/fixtures/mv/)の順に入れ、遅れた注文の数(order_id >= 2000000)をlate_orders、mv.cat_salesのsum(orders)をjoined_orders、その差をlost_ordersとして書き込んでください(保存先: /root/ch/mv/join.json)。

参考

ソーステーブルを作成して1回目の分を入れる

データベースmvとテーブルmv.ordersを作成してください。列はorder_id UInt64, ts DateTime, shop LowCardinality(String), product_id UInt32, user_id UInt32, qty UInt32, price UInt32の順序、エンジンはMergeTree、ORDER BY (shop, ts)です。次に/opt/lab/fixtures/mv/orders_1.sqlを1回だけ実行して、20万件を入れてください。

CREATE DATABASEとCREATE TABLEのあと、clickhouse-client --queries-fileでフィクスチャを渡します。フィクスチャはハッシュで作ったデータなので、何回実行しても同じ行ができますが、MergeTreeは重複を防がないので、2回入れると40万件になります。

ターゲットテーブルとマテリアライズドビューを作成する

ターゲットテーブルmv.daily_sales (day Date, shop LowCardinality(String), orders UInt64, revenue UInt64)をSummingMergeTree、ORDER BY (shop, day)で作成し、マテリアライズドビューmv.daily_sales_mvをTO mv.daily_salesで作成してください。ビューのSELECTはtoDate(ts) AS day, shop, count() AS orders, sum(qty * price) AS revenue FROM mv.orders GROUP BY day, shopです。作成した直後に、mv.daily_salesの行数を確認してみてください。

ビューは結果を直接保存せず、TOで指定したテーブルに書きます。ターゲットテーブルのソートキーをビューのGROUP BYに合わせておかないと、マージのときに同じ(shop, day)がまとまりません。作ったばかりのビューのターゲットテーブルが空である理由が、このモジュールの最初の教訓です。

ビューが見たものとソースを比べる

/opt/lab/fixtures/mv/orders_2.sqlを1回入れた直後に、mv.ordersの行数をsource_orders、mv.daily_salesのsum(orders)をview_ordersとして書き込んでください(保存先: /root/ch/mv/gap.json)。

ビューは、INSERTブロックを入力として受け取るトリガーです。2回目の分のINSERTはビューがあるときに入り、1回目の分はビューができる前に入りました。2つの数値の差が、そのままバックフィルすべき量です。

抜けた区間だけをバックフィルする

ビューが見なかった1回目の分(9月1–15日)をINSERT INTO mv.daily_sales SELECT ...でバックフィルして、mv.daily_salesを(shop, day)ごとに足した値が、ソース全体を集計し直した値と列ごとに一致するようにしてください。

ビューのSELECTをそのまま使い、ソースにWHEREで区間をかけるのが核心です。ビューがすでに入れた日まで入れ直すと、SummingMergeTreeがその日を2回足します。境界は、2回目の分が始まる時刻です。

マージ前でも正しくなるようにターゲットテーブルを読む

mv.daily_salesからshop = 'shop-3'の日付ごとの注文数・売上を求めるクエリを書いてください(/root/ch/mv/q_daily.sql)。結果の列はday, orders, revenueで、日付順です。ターゲットテーブルだけを読む必要があります。

ビューのGROUP BYは、ブロック1つの中でだけまとめます。そのため、マージが終わる前は、同じ(shop, day)が複数行あります。クエリでもう一度sumしてGROUP BY dayすれば、マージのタイミングに関係なく、同じ答えが出ます。

足せない値は状態として持つ

mv.shop_stats (shop LowCardinality(String), day Date, orders AggregateFunction(count), buyers AggregateFunction(uniqExact, UInt32))をAggregatingMergeTree、ORDER BY (shop, day)で、ビューmv.shop_stats_mvをTO mv.shop_statsで(countState() AS orders, uniqExactState(user_id) AS buyers、GROUP BY shop, day)作成し、すでにあるソースを1回バックフィルしてください。そして、店舗ごとの1か月のユニーク購入者をmv.shop_statsから求めるクエリを書いてください(/root/ch/mv/q_buyers.sql)。列はshop, buyersで、店舗順です。

毎日のユニーク購入者数を足すと、複数の日に買った人が重複して数えられます。-Stateは、結果の代わりにまとめられる中間状態を保存し、クエリ時に同じ関数の-Mergeでまとめます。バックフィルを2回行うと、countMergeが2倍になります。

Nullテーブルから始まるカスケードビュー

mv.ordersの列の後ろにstatus LowCardinality(String)を加えたテーブルmv.rawをENGINE = Nullで作成し、status = 'paid'の行のソースの7つの列だけをmv.ordersに送るビューmv.raw_mvを作成してください。次に/opt/lab/fixtures/mv/raw.sqlを1回だけ入れてください。

Nullエンジンのテーブルは何も保存しませんが、INSERTは受け取ってビューを起こします。raw_mvがmv.ordersに書いたブロックは、さらにdaily_sales_mvとshop_stats_mvを起こします。1回のINSERTが3つのテーブルに届きます。mv.rawをSELECTすると、いつも0行です。

JOINビューは右側のテーブルの変化を知らない

mv.products (product_id UInt32, category LowCardinality(String))(MergeTree、ORDER BY product_id)、mv.cat_sales (category LowCardinality(String), orders UInt64)(SummingMergeTree、ORDER BY category)、そしてmv.orders AS o INNER JOIN mv.products AS p ON o.product_id = p.product_idでカテゴリごとのcount() AS ordersをmv.cat_salesに送るビューmv.cat_sales_mvを作成してください。次に、/opt/lab/fixtures/mv/のproducts.sql→late.sql→products_new.sqlの順に入れ、遅れた注文の数(order_id >= 2000000)をlate_orders、mv.cat_salesのsum(orders)をjoined_orders、差をlost_ordersとして書き込んでください(保存先: /root/ch/mv/join.json)。

JOINのあるビューで、トリガーになるのは一番左のテーブル(ソース)のINSERTだけで、右側のテーブルは、その瞬間の内容をまるごと読むだけです。遅れた注文が入った瞬間に商品テーブルになかった商品の注文は、INNER JOINで捨てられ、商品があとから登録されても復活しません。