Databricks SQL での ETL
大量のデータを扱う場合、データセット全体を再処理するのではなく、新規レコードと変更されたレコードのみを処理できるパイプラインが必要です。これは増分ETLと呼ばれます。Databricks SQLでは、プロシージャル コードを記述したり手動更新をスケジュールしたりせずに、ストリーミング テーブルとマテリアライズドビューを使用して増分ETLパイプラインを構築できます。
このチュートリアルでは、製品の変更を時系列で追跡するという一般的なパターンについて説明します。ソーステーブルを作成し、変更イベントを取り込み、各製品の完全な履歴を保持するディメンションテーブルを構築し、最近のレコードをインクリメンタルに上書きするファクトテーブルを維持し、その上に集計レポートレイヤーを追加します。ディメンションテーブルに AUTO CDC を使用し、ファクトテーブルに REPLACE WHERE フローを使用します。
このチュートリアルの重要な機能はAUTO CDCです。従来のデータウェアハウスでは、ターゲットテーブルへの挿入、更新、削除イベントを調整するために複雑MERGE INTOステートメントを記述します。この方法は、特にイベントが順不同で発生する場合に、エラーが発生しやすい。AUTO CDCこれを処理してくれます。ビジネスキー、シーケンス列、そしてSCDタイプ1（最新の値のみ）かSCDタイプ2（完全な履歴）かを指定すると、Databricksが適切なマージロジックを自動的に適用します。CDCの概要については、 「AUTO CDC APIs : パイプラインを使用した変更データ キャプチャの簡素化」を参照してください。
このチュートリアルを終えると、以下のことができるようになります。
- 変更データフィードを使用して変更を追跡するソーステーブルを作成しました。
- CDCイベントストリームを理解するために、生の変更データを検査した。
- これらのイベントからSCDタイプ2ディメンションテーブルを作成するために
AUTO CDCを使用しました。 - 削除イベントをパイプラインを通して段階的に処理しました。
- REPLACE WHEREフローを使用して最近の行を増分的に上書きするファクトテーブルを構築しました。
- 集計レポートを段階的に維持するマテリアライズドビューを作成しました。
SCHEDULE REFRESH EVERY 1 DAYを設定することで、変更がパイプライン全体に自動的に反映されるようになります。
要件
このチュートリアルを完了するには、以下の条件を満たす必要があります。
- Unity Catalogが有効になっているDatabricksワークスペース。
- SQLウェアハウス(サーバレスまたはプロ)。
- コンピュート リソースを作成する権限、またはコンピュート リソースにアクセスする権限を持っています。
- アカウントのサーバーレスコンピュートが有効になります。 地域限定で利用可能な機能については、こちらをご覧ください。
ステップ 1: カタログとスキーマをセットアップする
Databricks SQLエディタを開き、作業用カタログとスキーマを設定します。選択したカタログとスキーマに対してUSE権限を持っている必要があります。
USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;
ステップ 2: ソーステーブルを作成してデータをロードする
products テーブルを、Databricksでのチェンジデータフィードの使用 (CDF) を有効にして作成します。CDF は、Delta Lake の機能であり、挿入、更新、削除のすべてをクエリ可能な変更ログとして記録します。これは、トランザクションソースシステムからのCDCストリームに似ていますが、変更は外部ログからではなく、Deltaテーブル内で直接キャプチャされます。ここではCDFを使用して、下流のパイプラインが使用する変更イベントを生成します。
-
テーブルを作成し、初期レコードをロードします。
SQLCREATE OR REPLACE TABLE products (
product_id INT,
product_name STRING,
category STRING,
warehouse STRING
)
TBLPROPERTIES (delta.enableChangeDataFeed = true);
INSERT INTO products VALUES
(1, 'Spoon', 'Cutlery', 'Seattle'),
(2, 'Fork', 'Cutlery', 'Portland'),
(3, 'Knife', 'Cutlery', 'Denver'),
(4, 'Chair', 'Furniture', 'Austin'),
(5, 'Table', 'Furniture', 'Chicago'),
(6, 'Lamp', 'Lighting', 'Boston'),
(7, 'Mug', 'Kitchenware', 'Seattle'),
(8, 'Plate', 'Kitchenware', 'Atlanta'),
(9, 'Bowl', 'Kitchenware', 'Dallas'),
(10, 'Glass', 'Kitchenware', 'Phoenix'); -
新しい製品、ウェアハウスの移動、カテゴリの再割り当てなどの上流の変更をシミュレートします。
SQLINSERT INTO products VALUES
(11, 'Napkin', 'Dining', 'San Francisco'),
(12, 'Coaster', 'Dining', 'New York');
UPDATE products SET warehouse = 'Los Angeles' WHERE product_id = 1;
UPDATE products SET category = 'Dining' WHERE product_id = 2;
ステップ 3: 変更データフィードをクエリする
下流パイプラインを構築する前に、生の変更イベントを確認して、 AUTO CDCが何を処理するのかを理解しておくと役立ちます。table_changes()関数はCDFログを読み込み、キャプチャされたすべての操作とメタデータ列を返します。
SELECT
product_id, product_name, warehouse,
_change_type, _commit_version
FROM table_changes('products', 1)
ORDER BY _commit_version, product_id;
例えば、スプーンには3つのイベントがあります。 insert （シアトル）、 update_preimage （シアトル）、 update_postimage （ロサンゼルス）です。
一つの論理的な変更（例えば、スプーンを別の場所に移動させる）によって、複数のイベント、つまり前像と後像が生じることに注目してください。 従来のデータウェアハウスでは、 MERGEステートメントを記述して、これらのすべてのイベントをターゲットテーブルに統合し、挿入、更新、削除を個別のロジックで処理し、イベントが正しい順序で適用されるようにします。これはまさに、次のステップでAUTO CDCが排除する複雑さです。
ステップ 4: SCDタイプ 2 ディメンションを構築します AUTO CDC
ストリーミングテーブルはデータを段階的に処理します。更新のたびに、前回の実行以降に追加された行のみを読み込むため、データセット全体を再処理する必要はありません。そのため、大量のデータソースや頻繁に変化するデータソースに最適です。
AUTO CDC 、ストリーミング テーブルに変更データ キャプチャ処理を追加します。 挿入、更新、削除を手動で処理するMERGE INTOステートメントを記述する代わりに、ビジネスキーとシーケンス列を宣言し、Databricks に適切なロジックを適用させます。AUTO CDC順不同のイベントも自動的に処理します。これは、 MERGE INTOを使用して分散システムや重複タイムスタンプを持つバッチロードから到着するイベントを処理する場合によくある問題です。
以下のステートメントは、各製品の完全なバージョン履歴を保持するSCDタイプ2テーブルを作成します。各バージョンには__START_ATと__END_ATタイムスタンプが付与されます。__END_AT内のNULLは現在のバージョンを示します。
CREATE OR REFRESH STREAMING TABLE products_history
SCHEDULE REFRESH EVERY 1 DAY
FLOW AUTO CDC
FROM (
SELECT * FROM STREAM products WITH (readChangeFeed = true)
WHERE _change_type != 'update_preimage'
)
KEYS (product_id)
APPLY AS DELETE WHEN _change_type = 'delete'
SEQUENCE BY _commit_timestamp
COLUMNS * EXCEPT (_change_type, _commit_version, _commit_timestamp)
STORED AS SCD TYPE 2;
SCHEDULE REFRESH EVERY 1 DAY: 毎日定期的にテーブルを更新します。FLOW AUTO CDC: これは CDC フローとして宣言します。Databricksは、挿入、更新、削除のセマンティクスを自動的に適用します。WHERE _change_type != 'update_preimage'：プレイメージをスキップします。ポストイメージにはすでに更新された行が存在します。KEYS (product_id): ビジネスの鍵。同じキーを持つイベントは、バージョン管理された行に統合されます。APPLY AS DELETE WHEN _change_type = 'delete': 削除イベントが発生したときに、現在のバージョンを閉じます。これにより、削除イベントを識別する条件を定義できます。SEQUENCE BY _commit_timestampイベントの順序を確立します。順不同の到着にも正しく対応します。STORED AS SCD TYPE 2: 完全な履歴を保持します。AUTO CDCSCDタイプ1とSCDタイプ2の両方をサポートしています。
ディメンションテーブルを照会します。
SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
- スプーン：2種類。シアトル（閉鎖、
__END_ATセット）とロサンゼルス（現在、__END_AT = NULL）。 - フォーク：2つのバージョン。カトラリーカテゴリー（販売終了）とダイニングカテゴリー（販売中）。
- ナプキンとコースター：それぞれ1種類（新規追加、
__END_AT = NULL）。 - その他の製品：各1バージョン（
__END_AT = NULL）。
ステップ 5: パイプラインを介して削除を処理します
次に、ソーステーブルから2つの販売終了製品を削除することで、それらをシミュレートします。
DELETE FROM products WHERE product_id = 9;
DELETE FROM products WHERE product_id = 10;
これらの削除イベントはCDFログに記録されますが、ストリーミングテーブルにはまだ反映されていません。新しいイベントを処理するために、ストリーミングテーブルを更新します。
REFRESH STREAMING TABLE products_history;
ディメンションテーブルを照会して、削除が適用されたことを確認します。
SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
ORDER BY product_id, __START_AT;
ボウルとグラスは現在__END_ATが設定されており、販売終了としてマークされています。その他の既存製品はすべて変更ありません。ストリーミングテーブルは、前回の更新時の挿入および更新を再処理することなく、新しい削除イベントのみを処理しました。
ステップ 6: REPLACE WHEREフローでファクトテーブルを構築する
前のステップのディメンションテーブルは、各製品の現在および過去の状態を追跡するため、 AUTO CDC を使用します。ファクトテーブルのニーズは異なります。ファクトテーブルは、売上などのイベントを記録します。通常は、遅れて到着したレコードを修正するなど、最近の期間のみを再処理します。このパターンの場合は、増分 REPLACE WHERE フローを使用します。これは、述語に一致する行のみを上書きし、他のすべての行はそのまま残します。マテリアライズドビューとこの構成の完全な比較については、REPLACE WHERE とマテリアライズドビューの使用時期を参照してください。
-
product_salesテーブルを作成し、いくつかの販売イベントをロードします。その一部は最近のもので、一部は古いものです。SQLCREATE OR REPLACE TABLE product_sales (
sale_id INT,
product_id INT,
quantity INT,
sale_date DATE
);
INSERT INTO product_sales VALUES
(101, 1, 3, current_date()),
(102, 2, 1, current_date()),
(103, 4, 5, date_add(current_date(), -2)),
(104, 6, 2, date_add(current_date(), -10)); -
更新ごとに過去7日間の売上のみを再計算するファクトテーブルを作成します：
SQLCREATE OR REFRESH STREAMING TABLE product_sales_recent
SCHEDULE REFRESH EVERY 1 DAY
FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7) BY NAME
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales;FLOW REPLACE WHERE sale_date >= date_add(current_date(), -7): 更新のたびに、述語と一致するターゲット内の行を削除し、同じ範囲のソースクエリーを再計算して、結果を挿入します。7日間のウィンドウの外にある行は変更されません。SCHEDULE REFRESH EVERY 1 DAY：ディメンションテーブルと同じ日次スケジュールでテーブルを更新します。BY NAME: 必須。列を位置ではなく名前で一致させます。
10日前の売上は述語の範囲外であるため、フローはそれを読み込みません。過去7日間の売上のみが保持されます。
-
ファクトテーブルにクエリーを実行する：
SQLSELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;結果には、過去7日間の売上が3件含まれています。
sale_dateの値は、チュートリアルを実行する日に応じて異なります：販売ID
product_id
数量
販売日
101 (基礎)
1
3
今日
102
2
1
今日
103
4
5
2日前
販売 104 は、日付が7日のウィンドウの外にあるため、結果に含まれていません。
ステップ7: 集計マテリアライズドビューの作成
ソースの変更に合わせて常に最新の状態に保たれるディメンションテーブルができたので、その上にレポートレイヤーを追加できます。
AMATERIALIZEDビューは、コンピュート前のクエリ結果を物理テーブルとして保存します。 通常のビューは、読み取りのたびにクエリを再実行しますが、マテリアライズドビューは結果を永続化し、更新のたびに上流の変更によって影響を受ける行のみを再計算します。そのため、クエリのパフォーマンスが重要なダッシュボードやレポートに最適です。
CREATE OR REPLACE MATERIALIZED VIEW products_by_category
SCHEDULE REFRESH EVERY 1 DAY
AS
SELECT
category,
COUNT(*) AS active_products
FROM products_history
WHERE __END_AT IS NULL
GROUP BY category;
SCHEDULE REFRESH EVERY 1 DAY は、このビューが毎日更新されることを意味します。ストリーミングテーブルの同じスケジュールと組み合わせることで、ソーステーブルへの変更が更新サイクルごとにディメンションと集計にカスケードされる3段階のパイプラインができました。ディメンションまたは集計を手動で更新する必要はありません。
SELECT * FROM products_by_category ORDER BY active_products DESC;
ステップ 8: エンドツーエンドの連鎖を検証する
完全なパイプラインの連鎖を検証するために、ソーステーブルを変更し、昨日の遅延到着セールを記録します。
UPDATE products SET warehouse = 'Seattle' WHERE product_id = 3;
INSERT INTO product_sales VALUES (105, 3, 4, date_add(current_date(), -1));
ナイフはデンバーからシアトルに移動します。UPDATE はパイプライン全体のカスケードをTriggerし、3つのステージがどのように連携するかを示しています：
productsチェンジデータフィードに変更イベントを記録します。products_historyイベントを処理し、ナイフの新しいバージョンを追加します。products_by_category影響を受けるカトラリーの行のみを再計算します。
ファクトテーブルは独自のスケジュールで更新されます。次のスケジュールされた更新を待たずに今すぐ遅れて到着した売上を把握するには、ファクトテーブルを手動で更新します。
REFRESH STREAMING TABLE product_sales_recent;
フローはproduct_salesから過去7日間を再計算するため、売上105はproduct_sales_recentに表示されます。
確認する：
SELECT product_id, product_name, warehouse, __START_AT, __END_AT
FROM products_history
WHERE product_id = 3
ORDER BY __START_AT;
SELECT * FROM products_by_category ORDER BY active_products DESC;
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;
掃除
このチュートリアルで作成されたリソースをクリーンアップするには、次のSQLを使用します。
DROP MATERIALIZED VIEW IF EXISTS products_by_category;
DROP TABLE IF EXISTS product_sales_recent;
DROP TABLE IF EXISTS product_sales;
DROP TABLE IF EXISTS products_history;
DROP TABLE IF EXISTS products;