【Azure Databricks】Lakeflow Spark 宣言型パイプラインの基本を知ろう

1. はじめに
2.Lakeflow Spark 宣言型パイプラインとは？
3.パイプラインとは？
4.SQLでパイプラインを宣言する手順
5. まとめ
- この記事を読んだ方へのオススメコンテンツはこちら

1. はじめに

皆さんこんにちは。

今回はAzure Databricks Lakeflow Spark 宣言型パイプラインについて説明していきます。

この連載では、Azure DatabricksのLakeflow Spark 宣言型パイプラインの基本から実行手順について説明していきます。

第1回：Lakeflow Spark 宣言型パイプラインの基本を知ろう（今回）
第2回：Lakeflow Spark 宣言型のパイプラインを実装してみよう
第3回：Lakeflow Spark 宣言型パイプラインのデータの品質管理とは？

2.Lakeflow Spark 宣言型パイプラインとは？

Lakeflow Spark 宣言型パイプライン (SDP) は、Azure DatabricksにおけるDelta Live Tables Foundationの後継として進化した機能です。

SDPは、バッチ処理とストリーミング処理を単一のフレームワークに統合することで、データパイプラインの構築・管理・実行を容易にします。

データセット（テーブルやビュー）は、SQLまたはPythonを用いた宣言的な方法で定義することができ、システムがそれらの依存関係を自動的に解決します。

さらに、SDPはUnity CatalogおよびLakeflowと密に連携しており、データの一元管理およびセキュアな運用を実現します。

【メリット】
・処理ステップを自動的にオーケストレーションし、正しい順序と最適なパフォーマンスの保証
・宣言型のシンプルな構文により、SQLまたはPythonで容易に定義可能（複雑なロジック不要）
・SCD Type 1/Type 2をサポートし、Change Data Capture（CDC）の実装を簡素化
・差分処理に対応し、新規または変更されたデータのみを処理可能
⇒ 例えば、データ品質ルールを定義してデータの正確性や完全性を検証したり、
ダッシュボードでデータ品質の状況を監視したり、
問題が発生した場合にアラートを受け取ったりできます。

3.パイプラインとは？

Lakeflow Spark Declarative Pipelines（SDP）におけるパイプラインは、データ処理ワークフローを構成および実行するためのリソースです。
1つのパイプラインは、1つ以上のフロー、ストリーミングテーブル、マテリアライズドビュー、およびシンクを含むことができます。

※1 マテリアライズドビューは、クエリの実行結果を物理テーブルとして保存する仕組みであり、複雑な変換処理を事前に計算することでパフォーマンスの最適化を実現します。
リクエスト時に都度ロジックを実行するのではなく、システムが上流データ（upstream）を自動的に追跡し、変更が発生した際に再計算を行うことで、常に整合性が保たれた正確な状態を維持します。

※2 ストリーミングテーブルは、ストリーミングデータおよび増分処理に特化したDeltaテーブルの一種であり、大規模データセットに対して高効率な処理を実現します。
各更新時に全データを再スキャン・再計算するのではなく、新規または変更されたデータのみを対象として処理を行うことで、システムリソースの消費を抑えつつ、常に最新状態のデータを維持することが可能です。

4.SQLでパイプラインを宣言する手順

Lakeflow Spark 宣言型パイプラインのパイプラインを宣言すると、データに対して行う変換処理をPythonやSQLのソースファイルで定義することができます。

Databricks Notebook上で各タスクを個別に構築・検証することで、データの正確性を確認したうえで、最終的に本番のパイプラインへ組み込むことができます。
本シリーズ（全3回）の第1回では、Lakeflow Spark 宣言型パイプラインにおいて、実際のデプロイに進む前段階として、Notebook上でSQLを用いて処理ロジックを定義する方法について解説します。

新規のノートブックを作成する
①こちらの手順でワークスペースにアクセスします。
②Azure Databricksポータル画面で、サイドバーを展開します。
③+New > Notebook を選択します。

④作成したノートブックが開かれます。

Notebookの名前を「Lakeflow SDP」に変更し、言語としてSQLを選択します。

4-1.オブジェクト、またはストレージまたはDBFS内のファイルからパイプラインの宣言

【文法】
CREATE OR REFRESH STREAMING TABLE [テーブル名]
COMMENT “[コメント]”
AS SELECT * FROM read_files(‘[DBFSまたはストレージのパス]’,format => ‘[ファイル形式]’);

CREATE OR REFRESH STREAMING TABLE yellow_tripdata_raw
COMMENT 'The raw yellow tripdata dataset, ingested from dbfs.'
AS SELECT * FROM read_files(
'dbfs:/FileStore/yellow_tripdata_lakeflowSDP.parquet',
format => 'parquet'
);

CREATE OR REFRESH STREAMING TABLE yellow_tripdata_raw

COMMENT 'The raw yellow tripdata dataset, ingested from dbfs.'

AS SELECT * FROM read_files(

'dbfs:/FileStore/yellow_tripdata_lakeflowSDP.parquet',

format => 'parquet'

);

4-2.アップストリームデータセットからパイプラインを宣言する

【文法】
CREATE OR REFRESH STREAMING TABLE [テーブル名](
CONSTRAINT [制約名] EXPECT ([条件])
)
COMMENT “[コメント]”
AS SELECT
[カラム名] AS [新しいカラム名]
FROM STREAM(live.[参照するデータセット]);

CREATE OR REFRESH STREAMING TABLE yellow_tripdata_prepared(
CONSTRAINT valid_tollsAmount EXPECT (tollsAmount IS NOT NULL),
CONSTRAINT valid_passengerCount EXPECT (passengerCount > 0) ON VIOLATION FAIL UPDATE
)
COMMENT "yellow tripdata cleaned and prepared for analysis."
AS SELECT
  passengerCount AS Passenger_Count,
  tipAmount AS Creditcard_Amount,
  tripDistance AS Trip_Distance
FROM STREAM(live.yellow_tripdata_raw);