ファクトETLワークフローの構築
によって Lorenz Verzosa, Krishna Satyavarapu, Peyman Mohajerian, ジェシー・ヘラビ 、 ブライアン・スミス（Bryan Smith） による投稿
Archived. This article has not been updated since the publish date above. The dynamic nature of information means that previously accurate content can become outdated or even obsolete over time. Readers are advised to exercise due diligence and cross-check any information found in this blog post before making decisions or adopting any practices based on said information.
ディメンショナルモデリングは、分析用データウェアハウスを構築するための実績のあるアプローチです。多くの組織がDatabricksのような現代的なプラットフォームに移行していますが、これらの基本的な手法は依然として適用されています。
パート1では 、 我々は 次元スキーマを設計しました 。パート2では、 次元テーブルのためのETLパイプラインを構築しました 。そしてパート3では、事実テーブルのETLロジックを実装し、効率性と整合性を重視します。
最初のブログでは、以下に示すように、事実テーブル、FactInternetSalesを定義しました。次元テーブルと比較して、事実テーブルはレコードの長さが比較的短く、次元テーブルへの外部キー参照、事実の測定値、縮退次元フィールド、および1つのメタデータフィールドのみが存在します：
注：以下の例では、最初の投稿からCREATE TABLE文を変更して、外部キー定義を含めるようにしました。これらは別のALTER TABLE文で定義する代わりにです。また、縮退次元フィールドに主キー制約を追加して、その役割をこの事実テーブルでより明確にしました。
テーブルの定義自体は比較的シンプルですが、「LastModifiedDateTime」というメタデータフィールドについて少し掘り下げておく価値があります。ファクトテーブルは列数（フィールド数）としては比較的少ない一方で、行数（レコード数）が非常に多くなる傾向があります。ファクトテ ーブルには、日々大量に発生する業務処理の結果が記録されるため、数百万〜数十億件のデータを保持していることも珍しくありません。
そのため、毎回のETLサイクルで全件抽出・再ロードを行うのではなく、新規データや変更されたレコードのみに処理を限定するのが一般的です。
どのレコードを抽出対象とするかを特定する方法は、ソースシステムやその基盤によりさまざまですが、もっとも信頼性が高いのは業務システム側で変更データキャプチャ（CDC）が実装されている場合です。ただし、CDC機能がない場合には、レコード作成時や更新時に記録されるタイムスタンプを活用するケースが多くなります。この方法は完璧とは言えませんが、経験豊富なETL開発者なら誰しもが「現実的に最も有効な手段の一つ」と認識しているアプローチです。
※補足：Lakeflow Connectの導入により、リレーショナルデータベースに対するCDC（変更データキャプチャ）を実現する新たな選択肢が登場しています。本ブログ執筆時点ではプレビュー版ではあるものの、対応するRDBMSが今後さらに拡充されれば、効率的な増分抽出を行う手段として有望です。
本ファクトテーブルにおける「LastModifiedDateTime」フィールドは、業務システム側で記録されたタイムスタンプを保持します。データを抽出する際には、まずこのフィールドの最新 値をファクトテーブル内から取得し、それを起点として増分（デルタ）抽出を行っていきます。
我々のファクトETLの高レベルのワークフローは以下のように進行します：
このワークフローをよりわかりやすくするために、以下のセクションではその主要なフェーズごとに説明していきます。ディメンションETLに関する投稿とは異なり、 今回のワークフローでは、各ステップで最も実装しやすい言語に応じて、SQLとPythonを組み合わせてロジックを構築していきます。
このように言語を柔軟に使い分けられる点は、Databricksプラットフォームの大きな強みのひとつです。最初に「全てをSQLで書く」または「全てをPythonで書く」といった選択をする必要はなく、データエンジニアは実装中に自由に切り替えることができます。
このワークフローの最初の2ステップは、業務システムから新規および更新済みのデータを抽出することに焦点を当てています。
ステップ1では、まず既に記録されている LastModifiedDateTime の最新値を簡単に参照します。ファクトテーブルが初期状態で空の場合（初回実行時など）は、対象のソースシステムからすべての関連データを抽出できるよう、十分過去の日付をデフォルト値として定義します。
以下はその処理を記述するSQLの例です：
これで、その値を使用して運用システムから必要なデータを抽出することができます。このクエリにはかなりの詳細が含まれていますが、WHERE句に注目してください。ここでは、前のステップからの最後に観測されたタイムスタンプ値を使用して、新規または変更された個々のラインアイテム（または新規または変更された販売オーダーに関連するアイテム）を取得します：
以前と同様に、抽出したデータは、ワークフローの後続のステップに 進む前に、データエンジニアのみがアクセスできるステージングスキーマのテーブルに永続化されます。追加のデータクレンジングを行う必要がある場合は、今すぐ行うべきです。
データウェアハウスのETLサイクルでは、通常、次元ETLワークフローを実行し、その直後に事実ワークフローを実行します。このようにプロセスを整理することで、事実レコードを次元データに接続するために必要なすべての情報が整っていることをより確実に保証できます。ただし、新しい次元指向のデータが到着し、事実に関連するトランザクションレコードによって取り込まれる狭いウィンドウが存在します。そのウィンドウは、事実データの抽出を遅延させる全体的なETLサイクルの失敗がある場合には増加します。そして、もちろん、ソースシステムでの参照失敗があると、疑わしいデータがトランザクションレコードに現れることがあります。
この問題から自分たちを守るために、ステージングされた事実データに存在し、現在の（未期限）レコードセットには存在しないビジネスキー値を、特定のディメンションテーブルに挿入します。このアプローチは、事実テーブルが参照できるビジネス（自然）キーと代替キーを持つレコードを作成します。これらのレコードは、対象のディメンションがタイプ2のSCDである場合、遅延到着としてフラグ付けされ、次のETLサイクルで適切に更新できます。
まず始めに、ステージングデータ内の主要なビジネスフィールドのリストを作成します。ここでは、これらのフィールドを動的に識別することを可能にする厳格な命名規則を利用しています：
注：以下のコード例ではPythonに切り替えます。Databricksは複数の言語の使用をサポートしており、同じワークフロー内でも使用できます。この例では、PythonはSQLの概念と整合性を保ちつつ、少し柔軟性を提供してくれます。これにより、より伝統的なSQL開発者にもこのアプローチがアクセス可能になります。
他のビジネスキーから日付キーを分離したことに注意してください。それらには後で戻りますが、今のところ、このテーブルの非日付（その他）キーに焦点を当てましょう。
日付以外のビジネスキーごとに、フィールドとテーブルの命名規則を使用して、そのキーを保持するべき次元テーブルを特定し、左セミ結合（NOT IN()比較に似ていますが、必要に応じて複数列のマッチングをサポート）を実行して、ステージングテーブルのその列の値が次元テーブルにないものを特定します。一致しない値を見つけた場合、単純にそれを次元テーブルに挿入し、IsLateArrivingフィールドに適切な設定を行います：
このロジックは、事実レコードが有効なエントリにリンクされていることを確認したい場合、私たちの日付次元参照に対してうまく機能します。しかし、多くの下流のBIシステムは、日付次元が最初と最後の値の間の連続した、中断されない日付のシリーズを格納するロジックを実装します。テーブルの値の範囲より前または後の日付に遭遇した場合、欠けているメンバーを入力するだけでなく、連続した範囲を保持するために必要な追加の値を作成する必要があります。そのため、遅延到着日には少し異なるロジックが必要です：
DatabricksやSpark SQLをあまり使ったことがない場合、この最後のステップの中心となるクエリはおそらく見慣れないものでしょう。sequence()関数は、指定された開始と停止に基づいて値のシーケンスを作成します。結果は配列で、explode() 関数を使用して展開することで、配列の各要素が結果セットの行を形成します。そこから、必要な範囲と次元テーブルの内容を比較して、どの要素を挿入する必要があるかを特定します。その挿入により、我々の事実レコードが参照する何かを持つように、この次元にスマートキーとして代理キー値を実装します。
これで、ステージングテーブル内のすべてのビジネスキーが対応するディメンションのレコードにマッチすることを確信できるようになったので、ファクトテーブルへの公開を進めることができます。
このプロセスの最初のステップは、これらのビジネスキーの外部キー値を参照することです。これは単一の公開ステップの一部として行うことができますが、クエリに含まれる結合の大量により、このアプローチは維持が難しくなることがよくあります。このため、我々は効率は少し落ちますが、理解しやすく、変更しやすいアプローチを取ることがあります。つまり、ビジネスキーを一つずつ参照し、それらの値をステージングテーブルに追加するというアプローチです。
再び、我々は命名規則を利用してこのロジックをより直感的に実装しています。我々の日付次元は役割演じる次元であり、したがってより可変的な命名規則に従っているため、これらのビジネスキーに対しては少し異なるロジックを実装します。
この時点で、ステージングテーブルにはビジネスキーと代理キーの値、および測定値、退化次元フィールド、およびLastModifiedDateソースシステムから抽出された値が格納されています。公開をより管理しやすくするために、利用可能なフィールドをファクトテーブルでサポートされているフィールドと一致させるべきです。それを行うためには、ビジネスキーを削除する必要があります：
注釈: この ソース データフレームは 前のコードブロックで定義されています。
フィールドが一致しているので、公開ステップは簡単です。私たちは、ファクトレコードの一意の識別子として機能する退化次元フィールドに基づいて、ファクトテーブル内の既存のレコードと新規レコードをマッチさせ、必要に応じて値を更新または挿入します：
このブログシリーズが、Databricksプラットフォーム上でディメンショナルモデルを構築しようとしている方々にとって有益な情報を提供できたことを願っています。このデータモデリングアプローチとそれに関連するETLワークフローに経験のある多くの人々が、Databricksを馴染みやすく、アクセスしやすく、RDBMSプラットフォームで実装されていたものと比べて最小限の変更で長く確立されたパターンをサポートできると感じることでしょう。変更が生じる場所では、PythonとSQLを組み合わせてワークフローロジックを実装する能力など、データエンジニアがこれにより作業をより直接的に実装し、時間をかけてサポートすることが容易になると願っています。
Databricks SQLについて詳しく知りたい方は、私たちのウェブサイトをご覧いただくか、ドキュメンテーションをお読みください。また、Databricks SQLのプロダクトツアーもご覧いただけます。既存のウェアハウスを高性能でサーバーレスのデータウェアハウスに移行し、優れたユーザーエクスペリエンスと低コストを実現したい場合、Databricks SQLが解決策です。-無料でお試しいただけます。
ブログを購読して、最新の投稿を受信トレイにお届けします。