Databricks SQL の ETL

大量のデータを処理する場合は、データセット全体を再処理するのではなく、新しいレコードと変更されたレコードのみを処理できるパイプラインが必要です。これは増分 ETL と呼ばれます。 Databricks SQL では、手続き型コードを記述したり、手動更新をスケジュールしたりすることなく、ストリーミングテーブルと具体化されたビューを使用して、増分 ETL パイプラインを構築できます。

このチュートリアルでは、一般的なパターン (時間の経過に伴う製品の変化の追跡) について説明します。ソーステーブルを作成し、変更イベントを記録し、各製品の全履歴を保持するディメンションテーブルを作成し、最近のレコードを段階的に上書きするファクトテーブルを維持し、その上に集計レポートレイヤーを追加します。ディメンションテーブルには AUTO CDC を使い、ファクトテーブルにはREPLACE WHERE フローを使います。

このチュートリアルの主な機能は AUTO CDCです。従来のウェアハウスでは、挿入、更新、および削除イベントをターゲットテーブルに調整するための複雑な MERGE INTO ステートメントを記述します。この方法は、特にイベントが順不同で到着する場合に、エラーが発生しやすくなります。 AUTO CDC これを自動的に処理します。ビジネスキー、シーケンス列、SCD タイプ 1 (最新の値のみ) と SCD タイプ 2 (完全な履歴) のどちらを使用するかを宣言し、Azure Databricksは正しいマージロジックを自動的に適用します。 CDC の概要については、「 AUTO CDC API: パイプラインを使用して変更データキャプチャを簡略化する」を参照してください。

このチュートリアルの最後には、次の内容が含まれます。

変更データフィードを使用して変更を追跡するソーステーブルを作成しました。
CDC イベントストリームを理解するために生の変更データを検査しました。
これらのイベントから SCD Type 2 ディメンションテーブルを作成するために AUTO CDC 使用されます。
パイプラインを通じて段階的に処理された削除イベント。
REPLACE WHERE フローで最近の行を増分で上書きするファクトテーブルを作成しました。
集計レポートを増分的に保持する具体化されたビューを作成しました。
SCHEDULE REFRESH EVERY 1 DAY がパイプラインを通じて変更を自動的に反映するように設定されています。

必要条件

このチュートリアルを完了するには、次の要件を満たす必要があります。

Unity カタログが有効になっているAzure Databricks ワークスペース。
SQL ウェアハウス (サーバーレスまたはプロ)。
コンピューティングリソースを作成したり、コンピューティングリソースにアクセスしたりするためのアクセス許可を持っている。
アカウントに対して有効になっているサーバーレスコンピューティング。「利用可能なリージョンに制限がある機能」を参照してください。

手順 1: カタログとスキーマを設定する

Databricks SQL エディターを開き、作業カタログとスキーマを設定します。選択したカタログとスキーマを USE するアクセス許可が必要です。

USE CATALOG <your-catalog>;
USE SCHEMA <your-schema>;

手順 2: ソーステーブルを作成してデータを読み込む

products (CDF) を有効にしたテーブルを作成します。 CDF は、挿入、更新、削除のたびにクエリ可能な変更ログとして記録する Delta Lake 機能です。これは、トランザクションソースシステムの CDC ストリームに似ていますが、変更は外部ログではなく Delta テーブル内で直接キャプチャされる点が異なります。ここでは CDF を使用して、ダウンストリームパイプラインが使用する変更イベントを生成します。

テーブルを作成し、初期レコードを読み込みます。

CREATE 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');

新しい製品、倉庫の移動、カテゴリの再割り当てを含むアップストリームの変更をシミュレートします。

INSERT 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;

たとえば、Spoon には、 insert (シアトル)、 update_preimage (シアトル)、 update_postimage (ロサンゼルス) の 3 つのイベントがあります。

1 つの論理的な変更 (たとえば、Spoon を別の倉庫に移動するなど) では、複数のイベント (事前イメージとイメージ後) が生成されます。従来のウェアハウスでは、 MERGE ステートメントを記述して、これらのイベントをすべてターゲットテーブルに調整し、挿入、更新、削除を個別のロジックで処理し、イベントが正しい順序で適用されるようにします。次のステップでこのAUTO CDC複雑さを排除します。

手順 4: SCD タイプ 2 ディメンションを作成する `AUTO CDC`

ストリーミングテーブルは、データを段階的に処理します。更新のたびに、前回の実行以降の新しい行のみが読み取られるので、完全なデータセットを再処理する必要はありません。これにより、大量のソースや頻繁に変化するソースに適しています。

AUTO CDC は、ストリーミングテーブルの上に変更データキャプチャ処理を追加します。挿入、更新、削除を手動で処理する MERGE INTO ステートメントを記述する代わりに、ビジネスキーとシーケンス列を宣言し、Azure Databricks正しいロジックを適用できるようにします。 AUTO CDC また、順不同のイベントも自動的に処理されます。これは、 MERGE INTO を使用して分散システムから到着するイベントや、タイムスタンプが重複するバッチ読み込みを処理する場合に一般的な問題です。

次のステートメントでは、各製品の完全なバージョン履歴を保持する SCD Type 2 テーブルを作成します。各バージョンは __START_AT および __END_AT タイムスタンプを取得します。 NULLの__END_ATは、現在のバージョンをマークします。

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 フローとして宣言します。 Azure Databricksは、挿入、更新、および削除セマンティクスを自動的に適用します。
WHERE _change_type != 'update_preimage': プレビュー画像をスキップします。 postimage には、すでに更新された行が含まれています。
KEYS (product_id): ビジネスキー。同じキーを持つイベントは、バージョン管理された行にマージされます。
APPLY AS DELETE WHEN _change_type = 'delete': 削除イベントが到着すると、現在のバージョンが閉じられます。これにより、削除イベントを識別する条件を定義できます。
SEQUENCE BY _commit_timestamp: イベントの順序付けを確立します。注文外の到着を正しく処理します。
STORED AS SCD TYPE 2: 完全な履歴を保持します。 AUTO CDC は、SCD タイプ 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テーブルを作成し、最近のものも古いものも含めていくつかの売上イベントを読み込みます。

CREATE 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日間の売上のみを再計算するファクトテーブルを作成しましょう:
```
CREATE 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日間の売上のみが保持されます。
ファクトテーブルを検索:
```
SELECT sale_id, product_id, quantity, sale_date
FROM product_sales_recent
ORDER BY sale_date DESC, sale_id;
```
結果には過去7日間の3件の売上が含まれています。 sale_date値はチュートリアルを実施した日によって異なります:

sale_id product_id 量販売日

101 1 3 今日

102 2 1 今日

103 4 5 2 日前

売上 104 は、その日付が7日間の範囲外であるため、結果には含まれていません。

sale_id	product_id	量	販売日
101	1	3	今日
102	2	1	今日
103	4	5	2 日前

ステップ7: 集約されたマテリアライズドビューを作成する

ソースの変更が反映されたディメンションテーブルが作成されたので、レポートレイヤーを上に追加できます。

具体化されたビューは、事前に計算されたクエリ結果を物理テーブルとして格納します。クエリから読み取るたびにクエリを再実行する通常のビューとは異なり、具体化されたビューは結果を保持し、更新のたびにアップストリームの変更の影響を受ける行のみを再計算します。これにより、クエリのパフォーマンスが重要なダッシュボードやレポートに適しています。

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はパイプライン全体のカスケードを起動し、3段階がどのように連携して動作するかを示します。

products が変更イベントを変更データフィードに記録します。
products_history はイベントを処理し、Knife の新しいバージョンを追加します。
products_by_category は、影響を受けるカトラリー行のみを再計算します。

ファクトテーブルは独自のスケジュールで更新されます。次の予定更新を待つ代わりに、遅れて到着したセールを今すぐ取り込むには、ファクトテーブルを手動で更新してください:

REFRESH STREAMING TABLE product_sales_recent;

フローは product_salesから過去7日間を再計算するため、product_sales_recentに販売 105 が現れるようになりました。

確認：

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;

その他のリソース

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-10-16

Databricks SQL の ETL

必要条件

手順 1: カタログとスキーマを設定する

手順 2: ソース テーブルを作成してデータを読み込む

手順 3: 変更データ フィードのクエリを実行する

手順 4: SCD タイプ 2 ディメンションを作成する AUTO CDC

手順 5: パイプライン経由で削除を処理する

ステップ6:REPLACE WHERE フローでファクトテーブルを作成する

ステップ7: 集約されたマテリアライズドビューを作成する

ステップ8:エンドツーエンドのカスケードを確認する

クリーンアップ

その他のリソース

フィードバック

その他のリソース

手順 2: ソーステーブルを作成してデータを読み込む

手順 3: 変更データフィードのクエリを実行する

手順 4: SCD タイプ 2 ディメンションを作成する `AUTO CDC`