Databricksチュートリアル：エンドツーエンドのレイクハウス分析パイプラインの実行

[Tutorial: Run an end\-to\-end lakehouse analytics pipeline \| Databricks on AWS](https://docs.databricks.com/getting-started/lakehouse-e2e.html#) [2022/11/18時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

Databricksでは、データプロフェッショナルがETL(extract/transform/load)パイプラインをクイックに開発、デプロイできるようにプロダクションレベルのツールのスイートを提供しています。Unity Catalogを用いることで、データスチュワードは組織全体のユーザーに対して、セキュアなストレージ認証情報、外部ロケーション、データベースオブジェクトを設定することができます。Databricks SQLを用いることで、アナリストはプロダクションのETLワークロードで使用されているのと同じテーブルに対してSQLクエリーを実行することができ、大規模なリアルタイムBIを実現することができます。

本書の最後まで読み進めることで、以下のことに慣れ親しむことができます。

1. [Unity Catalogが有効化された計算クラスターの起動](#ステップ1-クラスターを作成する)
1. [Databricksノートブックの作成](#ステップ2-databricksノートブックを作成する)
1. [Unity Catalog外部ロケーションへのデータの読み書き](#ステップ3-unity-catalogによって管理される外部ロケーションでデータの読み書きを行う)
1. [Auto Loaderを用いたUnity Catalogテーブルへのインクリメンタルなデータ取り込みの設定](#ステップ4-unity-catalogにデータを取り込むauto-loaderを設定する)
1. [データを処理、クエリー、プレビューするためにノートブックのセルを実行](#ステップ5-データを処理操作する)
1. [Databricksジョブとしてノートブックをスケジューリング](#ステップ6-ジョブをスケジュールする)
1. [Databricks SQLからUnity Catalogテーブルにクエリー](#ステップ7-databricks-sqlからテーブルにクエリーする)

このチュートリアルでは、Unity Catalogが有効化されたクラスターで一般的なETLタスクを完了するために、インタラクティブなノートブックを使用します。Unity Catalogを使用していない場合は、[Databricksで初めてのETLワークロードを実行する](https://qiita.com/taka_yayoi/items/e27fd8b44308ea4ec8d8)をご覧ください。

# 要件

Databricksにログインし、Data Science & Engineeringワークスペースにいること。詳細については、[Get started: Free trial & setup](https://docs.databricks.com/getting-started/index.html)をご覧ください。

:::note
**注意**
クラスター作成権限を持っていない場合でも、[クラスターにアクセス](https://qiita.com/taka_yayoi/items/991aae376c089df58504#%E3%82%AF%E3%83%A9%E3%82%B9%E3%82%BF%E3%83%BC%E3%81%AE%E8%A1%A8%E7%A4%BA)できるのであれば、以下のほとんどのステップを完了することができます。

Databricks SQLワークスペースのみにアクセスできる場合には、[データアナリストとしてDatabricksを使い始める](https://qiita.com/taka_yayoi/items/8cda2e2c85ca0ed45286)をご覧ください。
:::

# ステップ1: クラスターを作成する

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

1. サイドバーの![](https://docs.databricks.com/_images/clusters-icon.png)**Compute**をクリックします。
1. Computeページで**Create Cluster**をクリックします。New Clusterページが表示されます。
1. クラスターのユニークな名称を指定します。
1. **Single node**ラジオボタンを選択します。
1. **Access mode**ドロップダウンで**Single user**を選択します。
1. **Single user access**フィールドにご自身のメールアドレスが表示されていることを確認します。
1. 使用する**Databricks runtime version**を選択します。Unity Catalogを使うには11.1以降が必要です。
1. **Create Cluster**をクリックします。

Databricksクラスターの詳細については、[クラスター](https://qiita.com/taka_yayoi/items/c5d99cd77fe4bfcf69f0)をご覧ください。

# ステップ2: Databricksノートブックを作成する

Databricksでインタラクティブなコードを記述、実行するにはノートブックを作成します。

1. サイドバーの![](https://docs.databricks.com/_images/create-icon.png)**New**をクリックし、**Notebook**をクリックします。
1. Create Notebookページで:
    - ノートブックのユニークな名称を指定します。
    - デフォルト言語が**Python**に設定されていることを確認します。
    - **Cluster**ドロップダウンからステップ1で作成したクラスターを選択します。
    - **Create**をクリックします。

一番初めが空のセルのノートブックが開きます。

ノートブックの作成、管理の詳細に関しては[ノートブックの管理](https://qiita.com/taka_yayoi/items/c306161906d6d34e8bd5)をご覧ください。

# ステップ3: Unity Catalogによって管理される外部ロケーションでデータの読み書きを行う

インクリメンタルなデータの取り込みには[Auto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)を使うことをお勧めします。Auto Loaderは、クラウドオブジェクトストレージに新規ファイルが到着すると自動で検知し、処理を行います。

外部ロケーションに対するセキュアなアクセスを管理するためにUnity Catalogを使うことができます。外部ロケーションに対して`READ FILES`権限を持つユーザーやサービスプリンシパルは、データを取り込むためにAuto Loaderを使うことができます。

通常は、外部ロケーションに到着するデータは他のシステムから書き込まれます。このデモでは、外部ロケーションにJSONファイルを書き込むことでデータの到着をシミュレートすることができます。

以下のコードをノートブックセルにコピーします。`catalog`の値を、`CREATE CATALOG`と`USE CATALOG`権限を持つカタログ名で置き換えます。`external_location`の値を`READ FILES`、`WRITE FILES`、`CREATE EXTERNAL TABLE`権限を持つ外部ロケーションのパスで置き換えます。

外部ロケーションを全体的なストレージコンテナとして定義することができますが、多くの場合コンテナ内にネストされたディレクトリをポイントします。

外部ロケーションパスの適切なフォーマットは`"s3://bucket-name/path/to/external_location"`となります。

```py:Python

 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コードでは、指定されたカタログにユニークなデータベースを作成し、指定された外部ロケーションにユニークなストレージロケーションを作成するために、あなたのメールアドレスを使用します。このセルを実行することで、このチュートリアルに関係するすべてのデータが削除され、冪等性を持ってこのサンプルを実行できる様になります。接続されたシステムからあなたのソースガイブリケーションへのデータバッチの到着をシミュレートするために使用するクラスが定義され、インスタンスが作成されます。

ノートブックの新規セルにこのコードをコピーし、環境を構築するために実行します。

:::note
**注意**
このコードで定義される変数は、既存のワークスペース資産やユーザーと競合するリスクを引き起こすことなしに、安全に実行される必要があります。制限のあるネットワークやストレージ権限によって、このコードの実行時にエラーが生じることがあります。これらの制限のトラブルシュートに関しては、ワークスペース管理者に問い合わせてください。
:::

```py: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回まで実行することができます。

```py:Python
RawData.land_batch()
```

# ステップ4: Unity Catalogにデータを取り込むAuto Loaderを設定する

[Delta Lake](https://docs.databricks.com/delta/index.html)にデータを格納することをお勧めします。Delta LakeはACIDトランザク書を提供するオープンソースのストレージレイヤーであり、データレイクハウスを実現するものです。Delta Lakeは、Databricksで作成されるテーブルのデフォルトフォーマットとなっています。

Unity Catalogテーブルにデータを取り込む様にAuto Loaderを設定するには、以下のコードをノートブックの空のセルにコピーアンドペーストします。

```py:Python
# Import functions
from pyspark.sql.functions import input_file_name, 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(file_path)
  .select("*", input_file_name().alias("source_file"), current_timestamp().alias("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .option("mergeSchema", "true")
  .toTable(table))
```

Auto Loaderの詳細に関しては、[Auto Loader](https://qiita.com/taka_yayoi/items/df143647dcf5942b51c6)をご覧ください。

Unity Catalogと構造化ストリーミングに関しては、[Using Unity Catalog with Structured Streaming](https://docs.databricks.com/structured-streaming/unity-catalog.html)をご覧ください。

# ステップ5: データを処理、操作する

ノートブックはセルごとにロジックを実行します。セルのロジックを実行するには以下のステップを踏みます:

1. 以前のステップで入力したセルを実行するには、セルを選択し**SHIFT+ENTER**を推します。
1. 作成したテーブルにクエリーを実行するには、空のセルに以下のコードをコピーアンドペーストし、**SHIFT+ENTER**を押してセルを実行します。

    ```py:Python
    df = spark.read.table(table_name)
    ```

1. データフレームのデータをプレビューするには、空のセルに以下のコードをコピーアンドペーストし、**SHIFT+ENTER**を押してセルを実行します。

    ```py:Python
    display(df)
    ```

データをインタラクティブに可視化する方法については、[Databricksにおけるデータの可視化](https://qiita.com/taka_yayoi/items/b3f7bd30907967afee17)をご覧ください。

# ステップ6: ジョブをスケジュールする

Databricksジョブのタスクに追加することで、Databricksノートブックをプロダクションのスクリプトとして実行することができます。このステップでは、手動で起動する新規ジョブを作成します。

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

1. ヘッダーバーの右側にある**Schedule**をクリックします。
1. **Job name**にユニークな名称を入力します。
1. **Manual**をクリックします。
1. **Cluster**ドロップダウンで、ステップ1で作成したクラスターを選択します。
1. **Create**をクリックします。
1. ウィンドウが表示されるので、**Run now**をクリックします。
1. ジョブ実行の結果を参照するには、**Last run**タイムスタンプの隣の![](https://docs.databricks.com/_images/external-link.png)アイコンをクリックします。

ジョブの詳細については、[Databricksにおけるジョブ管理](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)をご覧ください。

# ステップ7: Databricks SQLからテーブルにクエリーする

使用しているカタログに対する`USE CATALOG`権限、使用しているスキーマ(データベース)に対する`USE SCHEMA`権限、テーブルに対する`SELECT`権限を持っている人は誰でも、お好きなDatabricks APIを用いてテーブルのコンテンツにクエリーを行うことができます。

画面の左上にある `+` の上のペルソナスイッチャーを用いてDatabricks SQLのUIに切り替えることができます。ドロップダウンメニューから**SQL**を選択します。

Databricks SQLでクエリーを実行するには、稼働中のSQLウェアハウスにアクセスする必要があります。

このチュートリアルで作成したテーブルは`target_table`という名前です。最初に指定したカタログと`e2e_lakehouse_<your_username>`というパターンのデータベースを用いてクエリーを行うことができます。作成したデータオブジェクトを参照するために[Data Explorer](https://docs.databricks.com/data/index.html)を使うこともできます。

# さらなるインテグレーション

Databricksとデータエンジニアリングツールのインテグレーションに関しては以下を参照ください。

- [Connect your favorite IDE](https://docs.databricks.com/dev-tools/index-ide.html)
- [Use dbt with Databricks](https://docs.databricks.com/partners/prep/dbt.html)
- [Learn about the Databricks Command Line Interface \(CLI\)](https://docs.databricks.com/dev-tools/cli/index.html)
- [Databricks Terraformプロバイダー](https://qiita.com/taka_yayoi/items/691ee4909b0961de5ad1)

### Databricks 無料トライアル

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