[2025年1月版] Delta Live Tablesチュートリアルのウォークスルー

明けましておめでとうございます。今年もよろしくお願いします。今年初めの記事はDelta Live Tables(DLT)からです。

**Delta Live Tablesとは**

Databricksにおけるデータパイプライン開発・運用のフレームワークです。背後ではSparkの構造化ストリーミングやDelta Lakeが動いており、チェンジデータキャプチャやストリーミングを必要とするようなデータパイプラインを低工数で実装でき、運用のために必要なオーケストレーション、自動テスト、オートスケール、イベントログ管理などの機能も提供してます。

https://www.databricks.com/jp/product/delta-live-tables

で、タイトルに戻ります。一昨年にこちらを書きました。

https://qiita.com/taka_yayoi/items/1f542ed8f36efba4a285

この時点では、パイプラインの画面からしかパイプラインを実行できませんでした。何を言っているのかと思われるかもしれませんが、Delta Live Tablesにおけるパイプラインの実装はDatabricksノートブックで行います。リリース当時はノートブックでロジックを実装、パイプラインの画面で処理を実行という形になっており、画面の行ったり来たりが発生していて、正直イケていませんでした。その後こちらの機能改善がなされました。

https://qiita.com/taka_yayoi/items/9c746ccf931fe81bda18

これによって、ノートブックから直接パイプラインを実行できるようになり、QoLが上がった訳です。そして、[マニュアル](https://docs.databricks.com/ja/delta-live-tables/tutorial-pipelines.html)を改めて見てみるとチュートリアルの流れも更新されていました。ということで、改めてチュートリアルをウォークスルーします。

# [CSVファイルのダウンロード](https://docs.databricks.com/ja/delta-live-tables/tutorial-pipelines.html#step-0-download-data)

適当なノートブックを作成して、以下を実行します。カタログやスキーマ、ボリューム名はアクセス権のあるものに適宜変更してください。しかし、[`dbutils.fs.cp`](https://docs.databricks.com/ja/dev-tools/databricks-utils.html#dbutils-fs-cp)で、URLから[ボリューム](https://docs.databricks.com/ja/volumes/index.html)に直接コピーできるの知りませんでした。

```py
my_catalog = "users"
my_schema = "takaaki_yayoi"
my_volume = "dlt_tutorial"

spark.sql(f"CREATE SCHEMA IF NOT EXISTS {my_catalog}.{my_schema}")
spark.sql(f"CREATE VOLUME IF NOT EXISTS {my_catalog}.{my_schema}.{my_volume}")

volume_path = f"/Volumes/{my_catalog}/{my_schema}/{my_volume}/"
download_url = "https://health.data.ny.gov/api/views/jxy9-yhdk/rows.csv"
filename = "babynames.csv"

dbutils.fs.cp(download_url, volume_path + filename)
```

ファイルがボリュームに保存されました。

![Screenshot 2025-01-02 at 19.46.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/78ccb943-de60-21b2-a061-26e58f394662.png)

# [ステップ 1: パイプラインを作成する](https://docs.databricks.com/ja/delta-live-tables/tutorial-pipelines.html#step-1-create-a-pipeline)

この部分でも知らない機能がありました。昔はパイプラインを作成する前にパイプラインの実装のノートブックを準備しておいて、作成時に選択する必要がありました。今では、事前のノートブックの準備が不要になっています。

パイプライン作成画面で名前やモードを指定していき、**ソースコード**の部分は空でもパイプラインを作成できます。この場合、空のノートブックが作成されるので、そこにパイプラインの実装を記述する形になります。

Delta Live Tablesパイプラインを実装する際には、処理結果を保持する[マテリアライズドビュー](https://docs.databricks.com/ja/delta-live-tables/index.html#materialized-view)を作成します。**配信先**では、これらのマテリアライズドビューを保存するカタログとスキーマを指定します。

![Screenshot 2025-01-02 at 19.16.47.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/492b93e7-de03-a7bc-fbde-46de15463129.png)

処理の成功や失敗を通知することができますが、ここではスキップします。**Advanced**ではパイプラインのパラメーターをキーバリュー方式で指定します。これで、開発時と運用時で挙動を変更できるようにパイプラインをパラメーター化することができます。このパラメーターは後のステップで参照します。

![Screenshot 2025-01-02 at 19.26.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1131d380-a643-0429-dd52-4d6a62d0b13d.png)


今回は**ソースコード**でノートブックを指定していないので、作成ボタンを押すと以下のようなダイアログが表示されます。これも以前は無かったフローです。

![Screenshot 2025-01-02 at 19.18.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/652aa22a-6109-a077-94b2-cada7b81938f.png)

# [ステップ 2: Python または SQL を使用してノートブックでマテリアライズド ビューとストリーミング テーブルを宣言する](https://docs.databricks.com/ja/delta-live-tables/tutorial-pipelines.html#step-2-declare-materialized-views-and-streaming-tables-in-a-notebook-with-python-or-sql)

パイプラインが作成されたのに合わせて、実装用のノートブックが作成されます。こちらを編集してパイプラインを実装します。

![Screenshot 2025-01-02 at 19.18.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/93549fc0-0976-59da-6d52-459bf2ff10f7.png)

![Screenshot 2025-01-02 at 19.19.02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/800518b6-c109-f26f-b245-c757acd625fd.png)

ただ、この時点ではこのノートブックは単なるノートブックです。Delta Live Tablesとの連携機能を活用するには、画面右上の**接続**をクリックして、上のステップで作成したパイプラインを選択、**接続**をクリックします。

![Screenshot 2025-01-02 at 19.19.28.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bf68899c-ace4-8965-1dd9-879773db2598.png)

![Screenshot 2025-01-02 at 19.19.34.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fb4129cb-91bc-a516-c12a-d61f1e611a57.png)

すると、画面下に**DLT...** と言ったペインが表示されるようになります。DLTはDelta Live Tablesの略です。また、画面の右上のボタンも変化し、**検証**、**起動**というものに変わります。これはDLTパイプラインの検証や起動のためのボタンとなります。機能の詳細は[こちら](https://docs.databricks.com/ja/delta-live-tables/dlt-notebook-devex.html)をご覧ください。

![Screenshot 2025-01-02 at 19.19.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/74261b0b-8256-1b5b-3d90-aab0bace8677.png)

これで実装の準備ができたので、画面上のセルに以下のロジックを記載します。DLTではPythonあるいはSQLでパイプラインを実装できますが、以下ではPython(PySpark)で実装しています。詳細は[Python](https://docs.databricks.com/ja/delta-live-tables/python-dev.html)や[SQL](https://docs.databricks.com/ja/delta-live-tables/sql-dev.html)のマニュアルをご覧ください。

```py
# モジュールのインポート

import dlt
from pyspark.sql.functions import *

# パイプラインパラメータを変数に割り当てる

my_catalog = spark.conf.get("my_catalog")
my_schema = spark.conf.get("my_schema")
my_volume = spark.conf.get("my_volume")

# ソースデータのパスを定義する

volume_path = f"/Volumes/{my_catalog}/{my_schema}/{my_volume}/"

# ボリュームからデータを取り込むストリーミングテーブルを定義する

@dlt.table(
  comment="ニューヨークの人気のある赤ちゃんのファーストネーム。このデータはニューヨーク州保健局から取り込まれました。"
)
def baby_names_raw():
  df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "csv")
    .option("inferSchema", True)
    .option("header", True)
    .load(volume_path)
  )
  df_renamed_column = df.withColumnRenamed("First Name", "First_Name")
  return df_renamed_column

# データを検証し、列名を変更するマテリアライズドビューを定義する

@dlt.table(
  comment="分析のためにクリーンアップおよび準備されたニューヨークの人気のある赤ちゃんのファーストネームデータ。"
)
@dlt.expect("valid_first_name", "First_Name IS NOT NULL")
@dlt.expect_or_fail("valid_count", "Count > 0")
def baby_names_prepared():
  return (
    spark.read.table("LIVE.baby_names_raw")
      .withColumnRenamed("Year", "Year_Of_Birth")
      .select("Year_Of_Birth", "First_Name", "Count")
  )

# データのフィルタリング、集計、およびソートされたビューを持つマテリアライズドビューを定義する

@dlt.table(
  comment="2021年のニューヨークのトップ赤ちゃんの名前のカウントを要約したテーブル。"
)
def top_baby_names_2021():
  return (
    spark.read.table("LIVE.baby_names_prepared")
      .filter(expr("Year_Of_Birth == 2021"))
      .groupBy("First_Name")
      .agg(sum("Count").alias("Total_Count"))
      .sort(desc("Total_Count"))
      .limit(10)
  )
```

ここでいきなり実行せずに、ロジックの検証を行ってみます。画面右上の**検証**をクリックします。この場合、データの処理は行わず、パイプラインのロジックの検証のみを行います。開発過程では頻繁に使うことになると思います。

![Screenshot 2025-01-02 at 19.20.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b866507d-d924-36f7-09f4-94131a5fe153.png)
![Screenshot 2025-01-02 at 19.20.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ef84264b-296b-7b80-ce62-d8d0e564c21c.png)

ロジックに問題がなければ、**DLTグラフ**タブにはパイプラインの依存関係(リネージ)が表示されます。

![Screenshot 2025-01-02 at 19.27.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/74db181d-847f-d6e1-0051-76c8c6e711cb.png)

# [ステップ 3: パイプラインの更新を開始する](https://docs.databricks.com/ja/delta-live-tables/tutorial-pipelines.html#step-3-start-a-pipeline-update)

これでパイプラインの準備ができました。今度は**起動**ボタンを押して実際にパイプラインの処理を実行します。パイプラインが起動すると、**DLTグラフ**リアルタイムで進捗を確認することができます。

![Screenshot 2025-01-02 at 19.31.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0a9921c6-d48d-405e-d9a3-15a610370fe0.png)
![Screenshot 2025-01-02 at 19.31.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/988e41b6-c4e8-d9ec-0737-9403e56bce64.png)
![Screenshot 2025-01-02 at 19.31.57.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/73624e03-9158-0787-f0e2-7b15d70ef14f.png)

また、**DLTイベントログ**にはパイプラインのイベントログが表示されます。

![Screenshot 2025-01-02 at 20.06.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3a9906a5-6ee2-14bf-592b-7c6c1bdd4216.png)

全ての処理が完了したら、[カタログエクスプローラ](https://docs.databricks.com/ja/catalog-explorer/index.html)でテーブルを確認します。生データを保持するブロンズテーブル、クレンジング結果を保持するシルバーテーブル、集計結果を保持するゴールドテーブルが作成されていることを確認できます。これらのテーブルは[メダリオンアーキテクチャ](https://docs.databricks.com/ja/lakehouse/medallion.html)に即したものとなっています。

![Screenshot 2025-01-02 at 19.33.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9e9262d6-c148-64ec-20d9-893085d7a9f6.png)
![Screenshot 2025-01-02 at 19.34.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c0036370-8563-f4ce-210d-1e85d34f38c1.png)
![Screenshot 2025-01-02 at 19.34.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/78c8be87-ca09-9c07-fbff-ef72596f7d09.png)

カタログエクスプローラでも、これらのテーブルのリネージを確認することができます。

![Screenshot 2025-01-02 at 19.34.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0e1ed2f3-8613-b191-7652-c9857976a0ad.png)

最後になりますが、今回のウォークスルーはノートブック中心に行いましたが、パイプラインの実行、運用にフォーカスする場合には、[Delta Live TablesパイプラインのUI](https://docs.databricks.com/ja/delta-live-tables/dlt-notebook-devex.html#how-to-access-the-delta-live-tables-ui-from-the-notebook)を活用することをお勧めします。パイプラインの状況をより俯瞰できるようになっています。

![Screenshot 2025-01-02 at 20.12.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cf1bf5d6-5cca-6228-af02-e4c1d9474e76.png)


### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)