チュートリアル:ETL Lakeflow宣言型パイプラインを使用してパイプラインを構築する

Lakeflow宣言型パイプラインとAuto Loaderを使用して、データオーケストレーション用のETL (抽出、変換、読み込み) パイプラインを作成およびデプロイする方法を学習します。ETL パイプラインは、ソースシステムからデータを読み取り、データ品質チェックやレコード重複排除などの要件に基づいてそのデータを変換し、データウェアハウスやデータレイクなどのターゲットシステムにデータを書き込む手順を実装します。

このチュートリアルでは、Lakeflow宣言型パイプラインとAuto Loaderを使用し、次のことを行います。

生のソースデータをターゲットテーブルに取り込みます。
生のソースデータを変換し、変換されたデータを 2 つのターゲットマテリアライズドビューに書き込みます。
変換されたデータをクエリします。
Databricks ジョブを使用して ETL パイプラインを自動化します。

Lakeflow 宣言型パイプラインと Auto Loaderの詳細については、「Lakeflow 宣言型パイプライン」および「Auto Loaderとは」を参照してください。

必要条件

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

Databricks ワークスペースにログインします。
ワークスペースで Unity Catalog が有効になっていることを確認します。
サーバレスコンピュートをアカウントで有効にしてください。サーバレス Lakeflow 宣言型パイプラインは、すべてのワークスペースリージョンで使用できるわけではありません。利用可能なリージョンについては、「利用可能な地域が限られている機能」を参照してください。
コンピュートリソースを作成する権限、またはコンピュートリソースにアクセスする権限を持っている。
カタログに新しいスキーマを作成する権限がある。必要な権限は、 ALL PRIVILEGES または USE CATALOG および CREATE SCHEMAです。
既存のスキーマに新しいボリュームを作成するアクセス許可がある。必要な権限は、 ALL PRIVILEGES または USE SCHEMA および CREATE VOLUMEです。

データセットについて

この例で使用されているデータセットは、現代音楽トラックの機能とメタデータのコレクションである Million Song データセットのサブセットです。このデータセットは、Databricksワークスペースに含まれているサンプルデータセットに含まれています。

ステップ 1: パイプラインを作成する

まず、Lakeflow宣言型パイプラインで ETL パイプラインを作成します。 Lakeflow 宣言型パイプラインは、宣言型パイプライン構文を使用してノートブックまたはファイル ( ソースコード と呼ばれます) で定義された依存関係を解決することで Lakeflow パイプラインを作成します。各ソースコードファイルには 1 つの言語のみを含めることができますが、パイプラインには複数の言語固有のノートブックまたはファイルを追加できます。詳細については、「Lakeflow 宣言型パイプライン」を参照してください。

important

ソースコード フィールドを空白のままにして、ソースコード作成用のノートブックを自動的に作成および構成します。

このチュートリアルでは、サーバレスコンピュートと Unity Catalogを使用します。指定されていないすべての構成オプションについては、デフォルト設定を使用します。サーバレスコンピュートがワークスペースで有効になっていないか、サポートされていない場合は、デフォルトコンピュートの設定を使用して、記載されているとおりにチュートリアルを完了できます。デフォルトコンピュート設定を使用する場合は、 パイプライン作成 UIの 配信先 セクションで ストレージオプション に Unity Catalog を手動で選択する必要があります。

Lakeflow宣言型パイプラインで新しい ETL パイプラインを作成するには次の手順を実行します。

ワークスペースで、サイドバーの ジョブとパイプライン をクリックします。
[ 新規 ] で、[ ETL パイプライン ] をクリックします。
パイプライン名 に、一意のパイプライン名を入力します。
サーバレス チェックボックスを選択します。
公開先 で、テーブルが公開される Unity Catalog の場所を構成するには、既存の カタログ を選択し、 スキーマ に新しい名前を書き込み、カタログに新しいスキーマを作成します。
作成をクリックします。

新しいパイプラインのパイプライン UI が表示されます。

ステップ 2: パイプラインを開発する

important

ノートブックには、1 つのプログラミング言語のみを含めることができます。パイプラインのソースコードノートブックで Python コードと SQL コードを混在させないでください。

この手順では、 Databricks ノートブックを使用して、Lakeflow宣言型パイプラインのソースコードを対話形式で開発および検証します。

このコードでは、インクリメンタルデータ取り込みに Auto Loader を使用します。 Auto Loader は、新しいファイルがクラウドオブジェクトストレージに到着すると、自動的に検出して処理します。詳細については、Auto Loaderとはを参照してください。

空白のソースコードノートブックが自動的に作成され、パイプライン用に構成されます。ノートブックは、ユーザー・ディレクトリー内の新しいディレクトリーに作成されます。新しいディレクトリとファイルの名前は、パイプラインの名前と一致します。たとえば、 /Users/someone@example.com/my_pipeline/my_pipelineです。

パイプラインを開発するときは、Python または SQL のいずれかを選択できます。両方の言語の例が含まれています。選択した言語に基づいて、デフォルトのノートブック言語を選択していることを確認します。Lakeflow宣言型パイプラインのコード開発に対するノートブックのサポートの詳細については、Lakeflow宣言型パイプラインでノートブックを使用してETLパイプラインを開発およびデバッグするを参照してください。

このノートブックにアクセスするためのリンクは、 パイプラインの詳細 パネルの ソースコード フィールドの下にあります。リンクをクリックしてノートブックを開き、次のステップに進みます。
右上の接続をクリックして、コンピュート設定メニューを開きます。
ステップ 1 で作成したパイプラインの名前にカーソルを合わせます。
接続をクリックします。
上部にあるノートブックのタイトルの横にあるノートブックのデフォルト言語(Python または SQL)を選択します。

次のコードをコピーして、ノートブックのセルに貼り付けます。

タブ :::タブ-item[Python]

Python
# Import modules
import dlt
from pyspark.sql.functions import *
from pyspark.sql.types import DoubleType, IntegerType, StringType, StructType, StructField

# Define the path to the source data
file_path = f"/databricks-datasets/songs/data-001/"

# Define a streaming table to ingest data from a volume
schema = StructType(
  [
    StructField("artist_id", StringType(), True),
    StructField("artist_lat", DoubleType(), True),
    StructField("artist_long", DoubleType(), True),
    StructField("artist_location", StringType(), True),
    StructField("artist_name", StringType(), True),
    StructField("duration", DoubleType(), True),
    StructField("end_of_fade_in", DoubleType(), True),
    StructField("key", IntegerType(), True),
    StructField("key_confidence", DoubleType(), True),
    StructField("loudness", DoubleType(), True),
    StructField("release", StringType(), True),
    StructField("song_hotnes", DoubleType(), True),
    StructField("song_id", StringType(), True),
    StructField("start_of_fade_out", DoubleType(), True),
    StructField("tempo", DoubleType(), True),
    StructField("time_signature", DoubleType(), True),
    StructField("time_signature_confidence", DoubleType(), True),
    StructField("title", StringType(), True),
    StructField("year", IntegerType(), True),
    StructField("partial_sequence", IntegerType(), True)
  ]
)

@dlt.table(
  comment="Raw data from a subset of the Million Song Dataset; a collection of features and metadata for contemporary music tracks."
)
def songs_raw():
  return (spark.readStream
    .format("cloudFiles")
    .schema(schema)
    .option("cloudFiles.format", "csv")
    .option("sep","\t")
    .option("inferSchema", True)
    .load(file_path))

# Define a materialized view that validates data and renames a column
@dlt.table(
  comment="Million Song Dataset with data cleaned and prepared for analysis."
)
@dlt.expect("valid_artist_name", "artist_name IS NOT NULL")
@dlt.expect("valid_title", "song_title IS NOT NULL")
@dlt.expect("valid_duration", "duration > 0")
def songs_prepared():
  return (
    spark.read.table("songs_raw")
      .withColumnRenamed("title", "song_title")
      .select("artist_id", "artist_name", "duration", "release", "tempo", "time_signature", "song_title", "year")
  )

# Define a materialized view that has a filtered, aggregated, and sorted view of the data
@dlt.table(
  comment="A table summarizing counts of songs released by the artists who released the most songs each year."
)
def top_artists_by_year():
  return (
    spark.read.table("songs_prepared")
      .filter(expr("year > 0"))
      .groupBy("artist_name", "year")
      .count().withColumnRenamed("count", "total_number_of_songs")
      .sort(desc("total_number_of_songs"), desc("year"))
  )

:::

タブ-item[sql]

SQL
-- Define a streaming table to ingest data from a volume
CREATE OR REFRESH STREAMING TABLE songs_raw
(
 artist_id STRING,
 artist_lat DOUBLE,
 artist_long DOUBLE,
 artist_location STRING,
 artist_name STRING,
 duration DOUBLE,
 end_of_fade_in DOUBLE,
 key INT,
 key_confidence DOUBLE,
 loudness DOUBLE,
 release STRING,
 song_hotnes DOUBLE,
 song_id STRING,
 start_of_fade_out DOUBLE,
 tempo DOUBLE,
 time_signature INT,
 time_signature_confidence DOUBLE,
 title STRING,
 year INT,
 partial_sequence STRING,
 value STRING
)
COMMENT "Raw data from a subset of the Million Song Dataset; a collection of features and metadata for contemporary music tracks."
AS SELECT *
FROM STREAM read_files(
'/databricks-datasets/songs/data-001/');

-- Define a materialized view that validates data and renames a column
CREATE OR REFRESH MATERIALIZED VIEW songs_prepared(
CONSTRAINT valid_artist_name EXPECT (artist_name IS NOT NULL),
CONSTRAINT valid_title EXPECT (song_title IS NOT NULL),
CONSTRAINT valid_duration EXPECT (duration > 0)
)
COMMENT "Million Song Dataset with data cleaned and prepared for analysis."
AS SELECT artist_id, artist_name, duration, release, tempo, time_signature, title AS song_title, year
FROM songs_raw;

-- Define a materialized view that has a filtered, aggregated, and sorted view of the data
CREATE OR REFRESH MATERIALIZED VIEW top_artists_by_year
COMMENT "A table summarizing counts of songs released by the artists each year, who released the most songs."
AS SELECT
 artist_name,
 year,
 COUNT(*) AS total_number_of_songs
FROM songs_prepared
WHERE year > 0
GROUP BY artist_name, year
ORDER BY total_number_of_songs DESC, year DESC

::: ::::

[ 開始 ] をクリックして、接続されたパイプラインの更新を開始します。

ステップ 3: 変換されたデータのクエリ

この手順では、ETL パイプラインで処理されたデータをクエリして、曲のデータを分析します。これらのクエリは、前の手順で作成した準備済みレコードを使用します。

まず、1990 年以降に毎年最も多くの曲をリリースしたアーティストを見つけるクエリを実行します。

サイドバーで、「 SQL エディター 」をクリックします。
新規タブアイコンをクリックし、メニューから 新規クエリーを作成 を選択します。

次の項目を入力します。

SQL
-- Which artists released the most songs each year in 1990 or later?
SELECT artist_name, total_number_of_songs, year
FROM <catalog>.<schema>.top_artists_by_year
WHERE year >= 1990
ORDER BY total_number_of_songs DESC, year DESC

<catalog> と <schema> を、テーブルが含まれているカタログとスキーマの名前に置き換えます。たとえば、 data_pipelines.songs_data.top_artists_by_year.

[選択項目を実行] をクリックします。

次に、4/4 ビートとダンサブルなテンポの曲を見つける別のクエリを実行します。

新しいタップアイコンをクリックし、メニューから 新しいクエリの作成 を選択します。
次のコードを入力します。
SQL
```
 -- Find songs with a 4/4 beat and danceable tempo
 SELECT artist_name, song_title, tempo
 FROM <catalog>.<schema>.songs_prepared
 WHERE time_signature = 4 AND tempo between 100 and 140;
```
<catalog> と <schema> を、テーブルが含まれているカタログとスキーマの名前に置き換えます。たとえば、 data_pipelines.songs_data.songs_prepared.
[選択項目を実行] をクリックします。

手順 4: パイプラインを実行するジョブを作成する

次に、 Databricks ジョブを使用してデータ取り込み、処理、分析のステップを自動化するワークフローを作成します。

ワークスペースで、サイドバーの ジョブとパイプライン をクリックします。
新規で、 ジョブ をクリックします。
タスクタイトルボックスで、 新しいジョブ <date and time> をジョブ名に置き換えます。たとえば、 Songs workflow.
[タスク名] に、最初のタスクの名前を入力します（例：ETL_songs_data）。
種類で パイプライン を選択します。
パイプライン で、ステップ 1 で作成したパイプラインを選択します。
作成をクリックします。
ワークフローを実行するには、 今すぐ実行 をクリックします。実行の詳細を表示するには、実行タブをクリックします。タスクをクリックして、タスク実行の詳細を表示します。
ワークフローの完了時に結果を表示するには、 成功した最新の実行に移動する か、ジョブ実行の開始時刻をクリックします。出力ページが表示され、クエリ結果が表示されます。

ジョブの実行の詳細については、「Lakeflowジョブのモニタリングと可観測性」を参照してください。

ステップ 5: パイプラインジョブをスケジュールする

ETL パイプラインをスケジュールに従って実行するには、次の手順を実行します。

ジョブと同じ Databricks ワークスペースの ジョブとパイプライン UI に移動します。
必要に応じて、[ ジョブ ] と [自分が所有] フィルターを選択します。
[名前] 列で、ジョブ名をクリックします。サイドパネルには ジョブの詳細 が表示されます。
スケジュールとトリガー パネルで トリガーの追加 をクリックし、 トリガー・タイプ で スケジュール済み を選択します。
期間、開始時刻、およびタイムゾーンを指定します。
[ 保存 ]をクリックします。

詳細情報

Lakeflow 宣言型パイプラインを使用したデータ処理パイプラインの詳細については、「Lakeflow 宣言型パイプライン」を参照してください。
Databricks ノートブックの詳細については、「Databricks ノートブック」を参照してください。
Lakeflowジョブの詳細については、「ジョブとは」を参照してください。
Delta Lake の詳細については、「 Databricks の Delta Lake とは」を参照してください。

必要条件​

データセットについて​

ステップ 1: パイプラインを作成する​

ステップ 2: パイプラインを開発する​

ステップ 3: 変換されたデータのクエリ​

手順 4: パイプラインを実行するジョブを作成する​

ステップ 5: パイプライン ジョブをスケジュールする​

詳細情報​