チュートリアル: エンドツーエンドの Lakehouse 分析パイプラインを実行する

[アーティクル]
08/27/2024

このチュートリアルでは、Azure Databricks レイクハウス用のエンドツーエンドの分析パイプラインを設定する方法について説明します。

重要

このチュートリアルでは、Unity Catalog 対応クラスターに対し、対話型ノートブックを使用して Python から一般的な ETL タスクを実行します。 Unity Catalog を使用していない場合は、「Azure Databricks で最初の ETL ワークロードを実行する」を参照してください。

このチュートリアルのタスク

この記事を読み終えるころには、次のような作業が快適に感じることでしょう。

Unity Catalog 対応のコンピューティングクラスターを起動する。
Databricks ノートブックを作成する。
Unity Catalog の外部の場所にデータを書き込んだりそこからデータを読み取ったりする。
自動ローダーを使用して Unity Catalog テーブルへの増分データインジェストを構成する。
ノートブックセルを実行してデータを処理、照会、プレビューする。
ノートブックを Databricks ジョブとしてスケジュールする。
Unity Catalog のテーブルに対して Databricks SQL からクエリを実行する。

Azure Databricks には、抽出、変換、読み込み (ETL) パイプラインをデータのプロフェッショナルがすばやく開発、デプロイできる運用環境向けのツールスイートが用意されています。データスチュワードは Unity Catalog を使用して、組織全体のユーザーを対象に、ストレージの資格情報、外部の場所、データベースオブジェクトを構成してセキュリティで保護できます。 Databricks SQL では、アナリストが運用環境の ETL ワークロードで使用されるのと同じテーブルに対して SQL クエリを実行できるため、リアルタイムのビジネスインテリジェンスを大規模に実現できます。

Delta Live Tables を使用して ETL パイプラインを構築することもできます。 Databricks は、運用 ETL パイプラインの構築、デプロイ、保守の複雑さを軽減するために Delta Live Tables を作成しました。「チュートリアル: 最初の Delta Live Tables パイプラインを実行する」を参照してください。

必要条件

Note

以降の手順の大半は、クラスターの制御権限がなくても、クラスターへのアクセス権があれば実行できます。

手順 1: クラスターを作成する

探索的データ分析やデータエンジニアリングを行うには、コマンドの実行に必要なコンピューティングリソースを提供するクラスターを作成します。

サイドバーで [コンピューティング] をクリックします。
サイドバーで、 [新規] をクリックし、[クラスター] を選びます。 [新しいクラスター/コンピューティング] ページが表示されます。
クラスターの一意の名前を指定します。
[Single node] (単一ノード) オプションボタンを選択します。
[アクセスモード] ボックスの一覧から [単一ユーザー] を選択します。
ご使用のメールアドレスが [単一ユーザー] フィールドに表示されていることを確認します。
Unity Catalog を使用するために必要な Databricks Runtime のバージョン (11.1 以降) を選択します。
[コンピューティングの作成] をクリックしてクラスターを作成します。

Databricks クラスターの詳細については、「コンピューティング」を参照してください。

手順 2: Databricks ノートブックを作成する

ワークスペースにノートブックを作成するには、サイドバーの [新規] をクリックしてから、[ノートブック] をクリックします。ワークスペースに空のノートブックが開きます。

ノートブックの作成と管理について詳しくは、「Notebooks を管理する」を参照してください。

手順 3: Unity Catalog によって管理される外部の場所にデータを書き込んだりそこからデータを読み込んだりする

Databricks では、自動ローダーを使用した増分データインジェストが推奨されています。自動ローダーは、クラウドオブジェクトストレージに到着した新しいファイルを自動的に検出して処理します。

Unity Catalog を使用して、外部の場所への安全なアクセスを管理します。外部の場所に対する READ FILES アクセス許可を持ったユーザーまたはサービスプリンシパルが自動ローダーを使用してデータを取り込むことができます。

通常、外部の場所に到着するデータは、他のシステムからの書き込みによるものです。このデモでは、外部の場所に JSON ファイルを出力することでデータの到着をシミュレートできます。

以下のコードをノートブックのセルにコピーしてください。 catalog の文字列値は、CREATE CATALOG と USE CATALOG のアクセス許可があるカタログの名前に置き換えます。 external_location の文字列値は、READ FILES、WRITE FILES、CREATE EXTERNAL TABLE の各アクセス許可がある外部の場所のパスに置き換えます。

外部の場所は、ストレージコンテナー全体として定義することもできますが、通常は、コンテナー内で入れ子になったディレクトリの場所を指定します。

外部の場所に使用する適切なパスの形式は "abfss://container_name@storage_account.dfs.core.windows.net/path/to/external_location" となります。


 external_location = "<your-external-location>"
 catalog = "<your-catalog>"

 dbutils.fs.put(f"{external_location}/filename.txt", "Hello world!", True)
 display(dbutils.fs.head(f"{external_location}/filename.txt"))
 dbutils.fs.rm(f"{external_location}/filename.txt")

 display(spark.sql(f"SHOW SCHEMAS IN {catalog}"))

このセルを実行すると、12 バイトという行と "Hello world!" という文字列が出力され、指定されたカタログに存在するすべてのデータベースが表示されます。このセルを実行できない場合は、使用中のワークスペースで Unity Catalog が有効になっていることを確認し、このチュートリアルを進めるうえで必要なアクセス許可をワークスペース管理者に要求してください。

以下の Python コードは、メールアドレスを使用して、指定されたカタログに一意のデータベースを作成し、指定された外部の場所に一意の保存場所を作成します。このセルを実行すると、このチュートリアルに関連するデータがすべて削除されるので、この例を何度実行しても同じ結果が得られます。接続されたシステムから外部のソースに到着する大量のデータをシミュレートするためのクラスが定義され、インスタンス化されます。

このコードをノートブックの新しいセルにコピーし、それを実行して環境を構成してください。

注意

このコードで定義されている変数を使用すれば、既存のワークスペース資産や他のユーザーと競合することなく安全にコードを実行できます。ネットワークまたはストレージのアクセス許可が制限されている場合は、このコードを実行するとエラーが発生します。ワークスペース管理者に連絡して、それらの制限のトラブルシューティングを行ってください。


from pyspark.sql.functions import col

# Set parameters for isolation in workspace and reset demo
username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first()[0]
database = f"{catalog}.e2e_lakehouse_{username}_db"
source = f"{external_location}/e2e-lakehouse-source"
table = f"{database}.target_table"
checkpoint_path = f"{external_location}/_checkpoint/e2e-lakehouse-demo"

spark.sql(f"SET c.username='{username}'")
spark.sql(f"SET c.database={database}")
spark.sql(f"SET c.source='{source}'")

spark.sql("DROP DATABASE IF EXISTS ${c.database} CASCADE")
spark.sql("CREATE DATABASE ${c.database}")
spark.sql("USE ${c.database}")

# Clear out data from previous demo execution
dbutils.fs.rm(source, True)
dbutils.fs.rm(checkpoint_path, True)

# Define a class to load batches of data to source
class LoadData:

    def __init__(self, source):
        self.source = source

    def get_date(self):
        try:
            df = spark.read.format("json").load(source)
        except:
            return "2016-01-01"
        batch_date = df.selectExpr("max(distinct(date(tpep_pickup_datetime))) + 1 day").first()[0]
        if batch_date.month == 3:
            raise Exception("Source data exhausted")
        return batch_date

    def get_batch(self, batch_date):
        return (
            spark.table("samples.nyctaxi.trips")
            .filter(col("tpep_pickup_datetime").cast("date") == batch_date)
        )

    def write_batch(self, batch):
        batch.write.format("json").mode("append").save(self.source)

    def land_batch(self):
        batch_date = self.get_date()
        batch = self.get_batch(batch_date)
        self.write_batch(batch)

RawData = LoadData(source)

後は以下のコードをセルにコピーして実行すれば、一括でデータを投入できます。新しいデータの到着をトリガーするために、このセルは最大 60 回手動で実行することができます。

RawData.land_batch()

手順 4: Unity Catalog にデータを取り込むように自動ローダーを構成する

Databricks はデータの格納に Delta Lake を使用することを推奨しています。 Delta Lake はオープンソースのストレージレイヤーです。ACID トランザクションを備え、データレイクハウスを実現します。 Databricks に作成されるテーブルの既定の形式は Delta Lake です。

Unity Catalog テーブルにデータを取り込むように自動ローダーを構成するには、ノートブックの空のセルに以下のコードをコピーして貼り付けてください。

# Import functions
from pyspark.sql.functions import col, current_timestamp

# Configure Auto Loader to ingest JSON data to a Delta table
(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source)
  .select("*", col("_metadata.source").alias("source_file"), current_timestamp().alias("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .option("mergeSchema", "true")
  .toTable(table))

自動ローダーの詳細については、「自動ローダー」を参照してください。

Unity Catalog での構造化ストリーミングについて詳しくは、「構造化ストリーミングで Unity Catalog を使用する」を参照してください。

手順 5: データを処理して対話的に操作する

ノートブックでは、ロジックがセル単位で実行されます。手順に従って、セル内のロジックを実行してください。

前の手順で完成したセルを実行するために、そのセルを選択し、Shift + Enter キーを押します。
作成されたテーブルに対してクエリを実行するために、以下のコードをコピーして空のセルに貼り付け、Shift + Enter キーを押してセルを実行します。
```
df = spark.read.table(table)
```
DataFrame 内のデータをプレビューするために、以下のコードをコピーして空のセルに貼り付け、Shift + Enter キーを押してセルを実行します。
```
display(df)
```

データを対話的に視覚化する方法について詳しくは、「Databricks ノートブックでの視覚化」を参照してください。

手順 6: ジョブをスケジュールする

Databricks ノートブックを運用スクリプトとして実行するには、それらをタスクとして Databricks ジョブに追加します。この手順では、手動でトリガーできる新しいジョブを作成します。

ノートブックをタスクとしてスケジュールするには:

ヘッダーバーの右側にある [スケジュール] をクリックします。
[ジョブ名] に一意の名前を入力します。
[手動] をクリックします。
[クラスター] ボックスの一覧から、手順 1. で作成したクラスターを選択します。
[作成] をクリックします。
表示されたウィンドウで、[今すぐ実行] をクリックします。
ジョブの実行結果を確認するには、前回実行タイムスタンプの横にあるアイコンをクリックします。

ジョブの詳細については、「Databricks ジョブとは」を参照してください。

手順 7: Databricks SQL からテーブルに対してクエリを実行する

現在のカタログとデータベースに対する USE CATALOG アクセス許可、現在のスキーマに対する USE SCHEMA アクセス許可、およびテーブルに対する SELECT アクセス許可があるユーザーはだれでも、任意の Databricks API からテーブルの内容を照会できます。

Databricks SQL でクエリを実行するには、実行中の SQL ウェアハウスへのアクセス権が必要です。

このチュートリアルで先ほど作成したテーブルには、target_table という名前が付いています。それに対してクエリを実行するためには、最初のセルで指定したカタログと e2e_lakehouse_<your-username> という形式のデータベースを使用します。作成されたデータオブジェクトは、カタログエクスプローラーを使用して確認できます。

その他の統合

Azure Databricks を使用したデータエンジニアリングのための統合とツールについては、以下の記事を参照してください。

次の方法で共有