【Azure Databricks】Delta Live Tablesのパイプラインを作成してみよう

1. はじめに
2.【おさらい】Delta Live Tablesパイプラインとは？
3.パイプラインを作成する
4.パイプラインを実行する
5. まとめ
- この記事を読んだ方へのオススメコンテンツはこちら

1. はじめに

皆さんこんにちは。

今回はAzure Databricks Delta Live Tablesのパイプラインの作成について説明していきます。

この連載では、Azure DatabricksのDelta Live Tablesの基本から実行手順について説明しています。

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

2.【おさらい】Delta Live Tablesパイプラインとは？

Delta Live Tablesパイプラインは、
Delta Live Tablesでデータ処理ワークフローを構成して実行するためのリソースです。

詳しくは前回の記事をご覧ください。

3.パイプラインを作成する

【準備】ノートブックを開く

〇既存のノートブックを開く
①こちらの手順でワークスペースにアクセスします。
②Azure Databricksポータル画面で、サイドバーを展開します。
③Workspace > Workspace > Users >自分のユーザー名 > 使用するノートブックを選択します。

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

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

3-1.Notebookにコマンドを入力する

【シナリオ】
①ストレージからのyellow trip datasetという生データを取り込み、tripdata_rawというテーブルを作成します。
②tripdata_rawテーブルをクレンジングし、 tripdata_preparedテーブルを作成します。
③tripdata_rawテーブルをクレンジングした、tripdata_cleanテーブルを作成し、
データ型/列名と品質の期待値を持つデータに加工します。
④ピックアップした場所ごとにグループ化し、通行料金の合計額を計算します。
さらに計算結果を降順に並び替え上位50件に絞り込んだtop_pagesテーブルを作成します。

【SQLの詳細】
①ストレージからのyellow trip datasetという生データを取り込み、tripdata_rawというテーブルを作成します。

CREATE OR REFRESH LIVE TABLE tripdata_raw
COMMENT"The raw yellow tripdata dataset,ingested from stgdtbricksprodjpeast999."
AS SELECT * FROM Parquet.`/mnt/delta-live-table/yellow_tripdata_delta_live_table_test.parquet`

CREATE OR REFRESH LIVE TABLE tripdata_raw

COMMENT"The raw yellow tripdata dataset,ingested from stgdtbricksprodjpeast999."

AS SELECT * FROM Parquet.`/mnt/delta-live-table/yellow_tripdata_delta_live_table_test.parquet`

②tripdata_rawテーブルをクレンジングし、 tripdata_preparedテーブルを作成します。

CREATE OR REFRESH LIVE TABLE tripdata_prepared(
CONSTRAINT vaild_tollsAmount EXPECT(Tolls_Amount IS NOT NULL),
CONSTRAINT vaild_passengerCount EXPECT(Passenger_Count &gt; 0)
)
COMMENT"yellow tripdata cleaned and prepared for analysis."
AS SELECT
 passenger_count AS Passenger_Count,
 tip_amount AS Creditcard_Amount,
 trip_distance AS Trip_Distance,
 tolls_amount AS Tolls_Amount
From live.tripdata_raw

CREATE OR REFRESH LIVE TABLE tripdata_prepared(

CONSTRAINT vaild_tollsAmount EXPECT(Tolls_Amount IS NOT NULL),

CONSTRAINT vaild_passengerCount EXPECT(Passenger_Count > 0)

)

COMMENT"yellow tripdata cleaned and prepared for analysis."

AS SELECT

passenger_count AS Passenger_Count,

tip_amount AS Creditcard_Amount,

trip_distance AS Trip_Distance,

tolls_amount AS Tolls_Amount

From live.tripdata_raw

③tripdata_rawテーブルをクレンジングした、tripdata_cleanテーブルを作成し、
データ型/列名と品質の期待値を持つデータに加工します。

CREATE LIVE TABLE tripdata_clean(
CONSTRAINT valid_tollsAmount EXPECT(Tolls_Amount IS NOT NULL),
CONSTRAINT valid_extra EXPECT(Extra &gt; 0)
)
COMMENT"yellow_tripdata with cleaned-up datatypes / column names and quality expectations."
TBLPROPERTIES("quality" = "silver")
AS SELECT
 CAST(puLocationID AS INT)AS PuLocationID,
 extra AS Extra,
 rateCodeID AS RateCodeID,
 tolls_amount AS Tolls_Amount
FROM live.tripdata_raw

CREATE LIVE TABLE tripdata_clean(

CONSTRAINT valid_tollsAmount EXPECT(Tolls_Amount IS NOT NULL),

CONSTRAINT valid_extra EXPECT(Extra > 0)

)

COMMENT"yellow_tripdata with cleaned-up datatypes / column names and quality expectations."

TBLPROPERTIES("quality" = "silver")

AS SELECT

CAST(puLocationID AS INT)AS PuLocationID,

extra AS Extra,

rateCodeID AS RateCodeID,

tolls_amount AS Tolls_Amount

FROM live.tripdata_raw

④ピックアップした場所ごとにグループ化し、通行料金の合計額を計算します。
さらに計算結果を降順に並び替え上位50件に絞り込んだtop_pagesテーブルを作成します。

CREATE LIVE TABLE top_pages
COMMENT"A list of the top 50 pages by passenger pay"
TBLPROPERTIES("quality" = "gold")
AS SELECT
 PuLocationId,
 SUM(Tolls_Amount) as Passenger_Pay
FROM live.tripdata_clean
GROUP BY PuLocationId
ORDER BY 2 DESC
LIMIT 50