Databricksで初めてのETLワークロードを実行する

[Run your first ETL workload on Databricks \| Databricks on AWS](https://docs.databricks.com/getting-started/etl-quick-start.html) [2022/5/27時点]の翻訳です。

> [Databricksクイックスタートガイド](https://qiita.com/taka_yayoi/items/125231c126a602693610)のコンテンツです。

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

Databricksでは、データのプロフェッショナルがETL(extract、transform、load)パイプラインをクイックに開発、デプロイできるようにプロダクションレディのツールのスイートを提供しています。

本書の最後までには、以下のことに習熟することになります。

1. [Databricksのall-purpose計算クラスターの起動](#ステップ1-クラスターを作成する)
1. [Databricksノートブックの作成](#ステップ2-databricksノートブックを作成する)
1. [Auto Loaderを用いたDelta Lakeへのインクリメンタルなデータ取り込みの設定](#ステップ3-delta-lakeにデータを取り込むauto-loaderを設定する)
1. [データを処理、クエリー、プレビューするためにノートブックセルを実行](#ステップ4-データを処理しインタラクションする)
1. [Databricksジョブとしてノートブックをスケジュール](#ステップ5-ジョブをスケジュールする)

このチュートリアルでは、PythonやScalaにおける一般的なETLタスクを完成させるためにインタラクティブなノートブックを使用します。

また、ETLパイプラインの構築にDelta Live Tablesを使用することもできます。プロダクションのETLパイプラインの構築、デプロイ、維持管理の複雑性を削減するために、DatabricksではDelta Live Tablesを作成しました。[Delta Live Tablesクイックスタート](https://qiita.com/taka_yayoi/items/7fe8ed2c2f95fd53cc3d)をご覧ください。

# 要件

- Databricksにログインし、Data Science & Engineeringワークスペースにアクセスしていること。詳細は[フリートライアルのサインアップ](https://qiita.com/taka_yayoi/items/fb4f57c069e1f272e88a)をご覧ください。
- [クラスターを作成できる権限](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%E3%82%A2%E3%82%AF%E3%82%BB%E3%82%B9%E6%A8%A9%E8%A8%AD%E5%AE%9A)を持っていること。

:::note info
**注意**
クラスターの管理権限を持っていない場合でも、[クラスターにアクセス](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ワークスペースにのみアクセスできる場合には、[Get started with Databricks as a data analyst](https://docs.databricks.com/getting-started/sql-quick-start.html)を参照ください。
:::

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

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

1. サイドバーの![](https://docs.databricks.com/_images/clusters-icon.png)**Compute**をクリックします。
1. クラスターページで、**Create Cluster**をクリックします。New Clusterページが表示されます。
1. クラスター固有の名前を指定し、他の値はデフォルトの状態のままとし、**Create Cluster**をクリックします。

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

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

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

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

一番上に空のセルがあるノートブックが開きます。

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

# ステップ3: Delta Lakeにデータを取り込むAuto Loaderを設定する

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

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

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

```py:Python
# Import a function
from pyspark.sql.functions import input_file_name, current_timestamp

# Define variables used in code below
file_path = "/databricks-datasets/structured-streaming/events"
username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first()[0]
table_name = f"{username}_etl_quickstart"
checkpoint_path = f"/tmp/{username}/_checkpoint/etl_quickstart"

# Clear out data from previous demo execution
spark.sql(f"DROP TABLE IF EXISTS {table_name}")
dbutils.fs.rm(checkpoint_path, True)

# 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)
  .toTable(table_name))
```

```scala:Scala
// Imports
import org.apache.spark.sql.functions.{input_file_name, current_timestamp}
import org.apache.spark.sql.streaming.Trigger
import spark.implicits._

// Define variables used in code below
val file_path = "/databricks-datasets/structured-streaming/events"
val username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first.get(0)
val table_name = s"${username}_etl_quickstart"
val checkpoint_path = s"/tmp/${username}/_checkpoint"

// Clear out data from previous demo execution
spark.sql(s"DROP TABLE IF EXISTS ${table_name}")
dbutils.fs.rm(checkpoint_path, true)

// 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.as("source_file"), current_timestamp.as("processing_time"))
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(Trigger.AvailableNow)
  .toTable(table_name)
```

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

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

# ステップ4: データを処理しインタラクションする

ノートブックはロジックをセル単位で実行します。お使いのセルでロジックを実行するには、以下を行います。

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

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

    ```scala:Scala
    val df = spark.read.table(table_name)
    ```

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

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

    ```scala:Scala
    display(df)
    ```

データの可視化を行うためのインタラクティブなオプションに関しては、[Databricksにおけるデータの可視化](https://qiita.com/taka_yayoi/items/36a307e79e9433121c38)をご覧ください。

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

Databricksジョブのタスクとして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)をご覧ください。

# その他のインテグレーション

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

- [Connect your favorite IDE](https://docs.databricks.com/dev-tools/index.html#use-an-ide)
- [Airflowによるデータパイプラインの依存関係の管理](https://qiita.com/taka_yayoi/items/5b5a3cb556acb57b67b1)
- [Use dbt with Databricks](https://docs.databricks.com/integrations/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)