Databricks上でのJAR(Javaなど)の実行方法

JavaをDatabricks上のSpark環境で実行する方法について説明します。

DatabricksのSparkランタイム(Databricks Runtime)は、

• 最新のSparkバージョン対応・最適化機能の享受
• OSS Sparkにはない高速化機能/拡張機能
• C++ベースのSparkエンジン(Photon)の高速化ランタイム
• インフラ管理・ソフトウェア/脆弱性管理などからの解放

などの特徴があり、コードはほぼそのままで、Databricks上に実行環境を移すだけで、実行速度の向上、それに伴う、費用削減などが見込めます(もちろんケースバイケースになります)。

Databricks上でのJARの実行方法

以下のステップになります。

1. JARファイルアップロードする
2. Jobの作成
3. Jobの実行と結果の確認

ここでは、前半でWeb UIを使用した方法に関して説明し、その後、CLI(コマンドライン)で実施する方法を説明します。

1. JARファイルのアップロード

JARをDatabricks上で実行するには、JARファイルをDBFS(DatabricksのRootストレージ、実体はWorkspaceを作成したときにS3, Blob Storage, GCS)にアップロードする必要があります。

Upload機能の有効化の確認

まず、Web UIからファイルをアップロード機能が有効化されているか確認します。
Admin権限を持つユーザーでDatabricksのWorkspaceにログインした後、以下の通り確認できます。

1. 左バーの下部にあるSettings(ギアのアイコン) > Admin Consoleをクリック
2. ページ遷移後に、上部にあるタブからWorkspace Settingsを選択
3. AdvancedセクションにあるDBFS File BrowserをONになっているか確認

以上で設定が完了しました。

JARのアップロード

Databricksの左メニューから

1. Dataを選択
2. 上部にあるDBFSボタンをクリック
3. その右側に現れるUploadボタンをクリック
4. ポップアップするWindow内でアップロードするディレクトリを指定
5. アップロードするファイルを選択する
6. Doneボタンをクリック

例えば、以下の通り設定します。

• アップロード先ディレクトリ: /FileStore/JAR_examples/
• アップロードするファイル: dataframe-example-1.0.JAR

アップロード後のファイルパスは以下の通りになります。

• dbfs:/FileStore/JAR_examples/dataframe_example_1_0.JAR

2. Jobの作成

Databricksの左メニューから　

1. Jobsをクリックして、Job一覧ページを開く
2. Create Jobボタンをクリック
3. Taskタブに現れるポップアップのWindow内にタスクの内容を設定
   • Task name: 任意の文字列(適宜)
   • Type: JAR
   • Main class: JARのメインクラス名
   • Depenent Libraries:
     • Addクリックして
     • Library Source: DBFS/S3
     • File Path: 先ほどアップロードしたJARファイルのパス(例: dbfs:/FileStore/JAR_examples/dataframe_example_1_0.JAR)
     • Addをクリック
4. Cluster: Editをクリックしてクラスタの内容を設定
5. Parameters: JARに渡すパラメータ(コマンドライン引数)をJSON形式で指定する (任意)
6. Createをクリック

3. Jobの実行と結果の確認

以下の通り、Jobの管理画面を開きます(Job作成した直後のページ)。

1. Databricksの左メニューからJobsをクリック
2. Job一覧から該当のJobをクリック
3. Job設定ページに以下の2つのタブがある
   • Runs: 実行結果の一覧
   • Tasks: Jobの設定(前節で実施した内容)

Jobの設定ページの右上にあるRun NowをクリックするとJobが実行開始されます。
実行が終了するとRunsタブにある一覧から結果が参照できます。Spark UI、標準/エラー出力なども参照可能です。

以上が基本的なJARの実行方法になります。

(補足1)クラスタについて

Jobを実行する際のクラスタは以下の2種類があります。

• Jobs Computeクラスタ(Jobsのデフォルト)

  • バッチ処理向けのクラスタタイプ
  • Job実行される毎に起動し、実行が終了した後すぐにシャットダウンされる
  • 起動に1-2数分かかる
  • 利用単価は安価(対All Purpose Compute)

• All Purpose Computeクラスタ(Notebook実行のデフォルト)

  • インタラクティブ作業向けのクラスタタイプ
  • 起動、シャットダウンはユーザーが実施する(idle時のオートシャットダウンの機能あり)
  • 連続起動しているため、すぐにJobの実行が可能
  • 利用単価は高価(対Jobs Compute)

本番運用時のバッチ処理ではJobs Computeクラスタが適していますが、開発時などの繰り返しJob実行する場合にはAll Purposeクラスタを使用することで効率化できます。

All Purposeクラスタの設定は、Databricksの左メニュー内のComputeで実施します。

(補足2) DBFSとS3 Bucketの対応について

上記で使用したDBFSストレージの実体はWorkspace作成時に登録したS3 Bucketです。
DBFS上のファイルパスとS3上のファイルパスの対応は以下の通りになっています。

システム	file path
DBFS上	dbfs:/FileStore/example/foobar.csv
S3上	s3://{DBFSのバケツ名}/tokyo-prod/{ワークスペースID}/FileStore/example/foobar.csv

また、SparkのコードからはDBFSのファイルパスはdbfs:を除いた/FileStore/example/foobar.csvでもアクセス可能です。

ファイル一覧参照については、DatabricksのUIから実施できます(左メニュー内のDataからDBFSをクリック)。

(補足3) Javaのコードについて

DatabricksのSparkを使用する際の注意点が以下のドキュメントから参照可能です。

• Job > Best Practice > JAR ジョブ
  • https://docs.microsoft.com/ja-jp/azure/databricks/jobs#jar-jobs

以下、Job実行確認のためのサンプルコードとJARファイルになります。

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;


public class DataframeApp {
  public static void main(String[] args){

    SparkSession spark = SparkSession
      .builder()
      .appName("Simple Daraframe Example")
      .getOrCreate();

    Dataset<Row> df = spark.read().format("csv")
      .option("inferSchema", "true")
      .option("header", "true")
      .load("/databricks-datasets/Rdatasets/data-001/csv/ggplot2/diamonds.csv");

    df.show();
    //spark.stop();
  }
}

• JARファイル: dataframe-example-1.0.jar

(補足4) コマンドラインからの実行について

上記で説明したJARの実行について、コマンドライン(CLI)からの実行も可能です。
以下にドキュメントがありますので、参照ください。

• ジョブ CLI

---

参考リンク

• Databricks Jobs