Databricksへのプロダクションワークロードの移行

[Migrate production workloads to Databricks \| Databricks on AWS](https://docs.databricks.com/migration/production.html) [2022/10/25時点]の翻訳です。

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

このガイドでは、他のプラットフォーム上のApache SparkからプロダクションジョブをどのようにDatabricks上のApache Sparkに移行するのかを説明します。

# コンセプト

[**Databricksジョブ**](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a)

バンドルしてサブミットできるコードの単一のユニットです。Databricksジョブは単一の`SparkContext`の[Sparkアプリケーション](https://spark.apache.org/docs/latest/submitting-applications)と同じものとなります。エントリーポイントはライブラリ(JAR、egg、wheelなど)やノートブックとなります。洗練されたリトライ、アラート機構を用いてDatabricksジョブをスケジュール実行することができます。ジョブ実行の主要なインタフェースはJobs [API](https://docs.databricks.com/dev-tools/api/latest/jobs.html)と[UI](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E4%BD%9C%E6%88%90)となります。

[**プール**](https://qiita.com/taka_yayoi/items/919acd7ef9decf0f61e6)

Databricksによって管理されるあなたのアカウント上のインスタンスのセットですが、アイドル状態の際にはDatabricksの課金は発生しません。プールで複数のジョブをサブミットすることで、ジョブがクイックに起動します。プールのインスタンスにガードレール(インスタンスタイプ、インスタンス制限など)とオートスケーリングのポリシーを設定することができます。プールは、他のSparkプラットフォームにおけるオートスケーリングクラスターと同じものです。

# 移行ステップ

このセクションでは、お使いのプロダクションジョブをDatabricksに移行するステップを説明します。

## ステップ1: プールの作成

[オートスケーリングプールを作成します](https://docs.databricks.com/clusters/instance-pools/create.html#instance-pools-create)。これは他のSparkプラットフォームでオートスケーリングクラスターを作るのと同じことです。他のプラットフォームにおいてオートスケーリングクラスターは数分あるいは数時間アイドル状態になったとしても、課金が発生します。Databricksは無料ではインスタンスプールを管理します。すなわち、これらのマシンが使用されていない場合には、Databricksに料金を支払う必要がありません。なお、クラウドプロバイダーの課金は発生します。Databricksでは、インスサンスでジョブが実行されている時*にのみ*課金が発生します。
![](https://docs.databricks.com/_images/pool.png)

重要な設定は以下の通りとなります。

- **Min Idle:** プールが管理するジョブで未使用のスタンバイ状態のインスタンスの数です。0に設定することができます。
- **Max Capacity:** これはオプションのフィールドです。すでにクラウドプロバイダーによる制限が設定されているのであれば、このフィールドは空のままでも構いません。さらに最大数の制限を設定したいのであれば、大量のジョブがプールを共有できる様に大きな値を設定してください。
- **Idle Instance Auto Termination:** 指定された時間アイドル状態であった場合、`Min Idle`を超えたインスタンスはクラウドプロバイダーに返却されます。値が大きいほど、多くのインスタンスがレディ状態で保持されるので、ジョブの起動が高速になります。

## ステップ2: プールでのジョブの実行

Jobs APIあるいはUIを用いてプールでジョブを実行することができます。それぞれのジョブに対してクラスターのスペックを指定する必要があります。ジョブが起動しようとすると、Databricksは自動でプールから新規クラスターを作成します。ジョブが完了するとクラスターは自動で停止します。ジョブが実行された時間分だけ課金されることになります。これは、Databricksでジョブを実行する際のコスト効率が最も高い方法となります。それぞれの新規クラスターは以下を持ちます:

- 他のSparkプラットフォームのSparkアプリケーションと同じ一つの`SparkContext`が関連づけられます。
- 一つのドライバーノードと指定された数のワーカーノード。単一のジョブに対して、ワーカー数の範囲を指定することができます。Databricksは当該ジョブが必要とするリソースに基づいて、単一のSparkジョブをオートスケールさせます。Databricksによる[ベンチマーク](https://www.databricks.com/blog/2018/05/02/introducing-databricks-optimized-auto-scaling.html)では、ジョブの性質によりますが、これによって最大30%クラウドコストを削減することができます。

プールでジョブを実行するには、API/CLI、Airflow、UIという三つの方法があります。

### API / CLI

1. [Databricks CLI](https://docs.databricks.com/dev-tools/cli/index.html)をダウンロードして設定します。
1. 一回、以下のコマンドをサブミットして実行します。このAPIは、ジョブ実行の進捗を追跡するために使用するURLを返却します。

    ```bash:Bash
    databricks runs submit --json
    
    {
      "run_name": "my spark job",
      "new_cluster": {
        "spark_version": "7.3.x-scala2.12",
    
        "instance_pool_id": "0313-121005-test123-pool-ABCD1234",
        "num_workers": 10
        },
        "libraries": [
        {
        "jar": "dbfs:/my-jar.jar"
        }
    
        ],
        "timeout_seconds": 3600,
        "spark_jar_task": {
        "main_class_name": "com.databricks.ComputeModels"
      }
    }
    ```

1. ジョブをスケジュールするには、以下のサンプルを使用します。このメカニズムを通じて作成されたジョブは[ジョブ一覧ページ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E5%8F%82%E7%85%A7)に表示されます。すべてのジョブ実行のステータスの参照に使用する`job_id`が返却されます。

    ```bash:Bash
    databricks jobs create --json
    
    {
      "name": "Nightly model training",
      "new_cluster": {
         "spark_version": "7.3.x-scala2.12",
         ...
         "instance_pool_id": "0313-121005-test123-pool-ABCD1234",
         "num_workers": 10
       },
       "libraries": [
         {
         "jar": "dbfs:/my-jar.jar"
         }
       ],
       "email_notifications": {
         "on_start": ["john@foo.com"],
         "on_success": ["sally@foo.com"],
         "on_failure": ["bob@foo.com"]
       },
       "timeout_seconds": 3600,
       "max_retries": 2,
       "schedule": {
       "quartz_cron_expression": "0 15 22 ? \* \*",
       "timezone_id": "America/Los_Angeles"
       },
       "spark_jar_task": {
         "main_class_name": "com.databricks.ComputeModels"
      }
    }
    ```

Sparkジョブをサブミットするのに[spark\-submit](https://spark.apache.org/docs/latest/submitting-applications#launching-applications-with-spark-submit)を使っている場合、以下のテーブルではspark-submitのパラメータがJobs APIで[新規ジョブ作成](https://docs.databricks.com/dev-tools/api/latest/jobs.html)オペレーション(`POST /jobs/create`)の引数にどのようにマッピングされるのかを示しています。

| spark-submitパラメーター | Databricksでどのように適用するのか |
|:--|:--|
| –class  | メインクラスの名前とパラメーターを指定するために[spark\_jar\_task structure](https://docs.databricks.com/dev-tools/api/latest/jobs.html)を使います。  |
| –jars  |  依存関係のリストを指定するために`libraries`を使います。 |
| –py-files  | Pythonジョブに対しては、[spark\_python\_task structure](https://docs.databricks.com/dev-tools/api/latest/jobs.html)を使います。eggやwheelの依存関係を指定するために`libraries`引数を使用することができます。  |
| –master  | クラウドでは、長時間処理を実行するマスターノードを管理する必要がありません。すべてのインスタンスとジョブはDatabricksサービスによって管理されます。このパラメーターは無視してください。  |
| –deploy-mode  | Databricksではこのパラメーターは無視してください。  |
| –conf  | [new\_cluster structure](https://docs.databricks.com/dev-tools/api/latest/jobs.html)では、`spark_conf`引数を使ってください。  |
| –num-executors  | [new\_cluster structure](https://docs.databricks.com/dev-tools/api/latest/jobs.html)では、`num_workers`引数を使ってください。また、範囲を指定するために`autoscale`オプションを使うこともできます(推奨)。  |
| –driver-memory, –driver-cores  | 必要とするドライバーメモリーとコアに基づいて、適切なインスタンスタイプを選んでください。プール作成時にドライバーのインスタンスタイプを指定するので、ジョブのサブミットの際にはこのパラメーターを無視してください。  |
| –executor-memory, –executor-cores  | 必要とするエグゼキューターメモリーとコアに基づいて、適切なインスタンスタイプを選んでください。プール作成時にエグゼキューターのインスタンスタイプを指定するので、ジョブのサブミットの際にはこのパラメーターを無視してください。  |
| –driver-class-path |  spark_conf引数では、`spark.driver.extraClassPath`に適切な値を設定してください。 |
| –driver-java-options  | spark_conf引数では、`spark.driver.extraJavaOptions`に適切な値を設定してください。  |
| –files  |  spark_conf引数では、`spark.files`に適切な値を設定してください。 |
| –name  | [submit job run request](https://docs.databricks.com/dev-tools/api/latest/jobs.html) (`POST /jobs/runs/submit`)では、`run_name`引数を使ってください。[create job request](https://docs.databricks.com/dev-tools/api/latest/jobs.html)(`POST /jobs/create`)では、`name`引数を使ってください。 |

### Airflow

DatabricksでジョブをサブミットするためにAirflowを使いたい場合、Databricksでは[Airflow operator](https://airflow.apache.org/docs/apache-airflow-providers-databricks/stable/operators.html)を提供しています。DatabricksのAirflowオペレーターは、DatabricksにジョブをサブミットするためにJobs APIの[Trigger a new job run](https://docs.databricks.com/dev-tools/api/latest/jobs.html)オペレーション(`POST /jobs/run-now`)をコールします。[Orchestrate Databricks jobs with Apache Airflow](https://docs.databricks.com/workflows/jobs/how-to-use-airflow-with-jobs.html)をご覧ください。

### UI

Databricksではジョブをサブミットしスケジュールするために、シンプルかつ直感的な使いやすいUIを提供しています。UIを用いてジョブを作成しサブミットするには、[ステップバイステップのガイド](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E3%81%AE%E4%BD%9C%E6%88%90)に従ってください。

## ステップ3: ジョブのトラブルシュート

Databricksでは、ジョブのトラブルシュートに役立つツールを数多く提供しています。

### ログとSpark UIへのアクセス

Databricksでは、それぞれのジョブ実行におけるすべてのSparkログやSpark UIにアクセスできるようにする完全マネージドのヒストリーサーバーを管理しています。これらには、ジョブ実行の詳細ページや[ジョブの実行ページ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#%E3%82%B8%E3%83%A7%E3%83%96%E5%AE%9F%E8%A1%8C%E3%81%AE%E5%8F%82%E7%85%A7)からアクセスすることができます。
![](https://docs.databricks.com/_images/job.png)

### ログの転送

お使いのクラウドストレージロケーションにクラスターログを転送することができます。選択したロケーションにログを転送するには、[new\_cluster structure](https://docs.databricks.com/dev-tools/api/latest/jobs.html)の`cluster_log_conf`パラメーターを使います。

### メトリクスの参照

ジョブの実行中、クラスターページに移動し、**Metrics**タブでライブのGangliaメトリクスを参照することができます。また、Databricksはこれらのメトリクスのスナップショットを15分ごとに作成して格納しますので、ジョブが完了した後でもこれらのメトリクスを参照することができます。メトリクスをご自身のメトリクスサーバーに送信するために、クラスターにカスタムエージェントをインストールすることができます。[パフォーマンス監視](https://qiita.com/taka_yayoi/items/991aae376c089df58504#%E3%83%91%E3%83%95%E3%82%A9%E3%83%BC%E3%83%9E%E3%83%B3%E3%82%B9%E7%9B%A3%E8%A6%96)をご覧ください。
![](https://docs.databricks.com/_images/metrics.png)

### アラートの設定

ジョブ失敗時にアラートを受け取れる様に、Jobs APIの[Create a new job](https://docs.databricks.com/dev-tools/api/latest/jobs.html)オペレーション(`POST /jobs/create`)で`email_notifications`を使います。

また、これらのメールのアラートをPagerDuty、Slackや他のモニタリングシステムに転送することもできます。

- [How to set up PagerDuty alerts with emails](https://www.pagerduty.com/docs/guides/email-integration-guide/)
- [How to set up Slack notification with emails](https://get.slack.help/hc/articles/206819278-Send-emails-to-Slack)

# FAQ

**プールなしにジョブを実行できますか？**

はい。プールはオプションです。新規クラスターでジョブを直接実行することができます。この様な場合、Databricksがクラウドプロバイダーに必要なインスタンスをリクエストすることでクラスターが作成されます。プールを用いることで、プールでインスタンスが利用できる場合には、クラスターの起動時間は30秒程度になります。

**ノートブックジョブとは何ですか？**

Databricksでは、JAR、Python、ノートブックといった異なるジョブタイプがあります。ノートブックジョブタイプは、指定されたノートブックの中でコードを実行します。

**JARジョブと比べた際、いつノートブックジョブを使うべきですか？**

JARジョブはspark-submitジョブと同じものです。JARを実行し、トラブルシュートの際にはログやSpark UIを参照することができます。ノートブックジョブは指定されたノートブックを実行します。ノートブックでライブラリをインポートし、ノートブックからライブラリをコールすることもできます。`main`エントリーポイントとしてノートブックジョブを使うことの利点は、ノートブックの出力領域でジョブの中間結果を容易にデバッグできるということです。[JARジョブ](https://qiita.com/taka_yayoi/items/b3275a1983c51a8bbe1a#jar%E3%82%B8%E3%83%A7%E3%83%96)をご覧ください。

**自分のHiveメタストアに接続できますか？**

はい、Databricksでは外部のHiveメタストアとGlueカタログをサポートしています。

- [External Apache Hive metastore](https://docs.databricks.com/data/metastores/external-hive-metastore.html#external-hive-metastore)
- [DatabricksランタイムのメタストアとしてAWS Glueデータカタログを使う](https://qiita.com/taka_yayoi/items/2cacbcaf9d70bfe80926)

### Databricks 無料トライアル

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