watsonx.data の WebコンソールからSpark job を実行してみた

はじめに

watsonx.data の Webコンソールから Spark アプリケーションを手動で実行する事ができます。マニュアルの記述のみでは判りにくい所があるので手順をご紹介します。

参照マニュアル

コンソールからSparkアプリケーションを送信する

事前準備

環境の準備

本記事の前提として下記の環境を準備しました。
・IBM Cloud Pak for Data 5.3.1 オンプレ環境
・watsonx.data 2.3.2
・Spark エンジン
・IBM Cloud Object Storage (ICOS)上のバケット

実行する Spark アプリケーションの準備

今回は下記の サンプル Spark アプリケーション (Pythonスクリプト)を実行します。
実質的な処理は、print でメッセージを1行表示するだけです。
ファイル名は "spark-sample.py" とします。

from pyspark.sql import SparkSession

def init_spark():

    spark = SparkSession.builder \
        .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions') \
        .enableHiveSupport() \
        .getOrCreate()
    return spark

def main():
    try:
        spark = init_spark()

    finally:
        spark.stop()

    print("This is a sample spark application")

if __name__ == '__main__':
    main()

実行する Spark アプリケーション をバケットにコピー

実行するアプリケーションのファイルは、予め ICOS 上のバケットにコピーしておきます。今回は "aws s3 cp" コマンドを使用して "spark-bucket0610" バケットにコピーしました。以下はコマンドの実行例です。

$ aws s3 cp spark-sample.py --endpoint-url https://s3.jp-tok.cloud-object-storage.appdomain.cloud s3://spark-bucket0610/spark-sample.py
upload: ./spark-sample.py to s3://spark-bucket0610/spark-sample.py

ZenApiKey の作成

参照マニュアル

認証用 API 鍵の生成
ZenApiKey許可トークンの生成
Spark アプリケーションを実行する際に、"ZenApiKey" と呼ばれるAPIキーを指定します。"ZenApiKey" を作成するために、まず 認証用 API 鍵の生成 を参照して "プラットフォームAPIキー" を生成してコピーしておきます。
次に下記のコマンドで "ZenApiKey" を作成します。

echo "<Software Hub のユーザー名>:<プラットフォームAPIキー>" | base64

以下は実行例です。

$ echo "cpadmin:LUHCD....I9bmU" | base64
Y3BhZG1pbjp.....Uk5Ym1VCg==

バケットの登録と Spark エンジンとの関連付け

watsonx.data の Webコンソールにログインして"インフラストラクチャー・マネージャー"を開きます。
ICOS 上に作成したバケット "spark-bucket0610" を"インフラストラクチャー・マネージャー"の「コンポーネントの追加」でストレージとして登録します。この時「カタログの関連付け」をチェックして、ストレージとカタログの関連付けを実施します。
その後、Spark エンジンとカタログを関連付けしておきます。
これらを実施した後の"インフラストラクチャー・マネージャー"は下記のようになります。

Spark アプリケーションの実行

"インフラストラクチャー・マネージャー"上の Spark エンジンをクリックして、次の画面で「アプリケーション」タブを開きます。

右上にある「アプリケーションの作成」をクリックして、アプリケーションの情報を入力します。
「アプリケーション・タイプ」で"Python"をチェックします。
「アプリケーション・パス」を "s3a://<バケット名>/<ファイル名>" の形式で入力します。今回は下記の内容になります。

s3a://spark-bucket0610/spark-sample.py

「アプリケーション・パス」を入力すると「アプリケーション名」が自動的に入力されます。(変更可能)
他の入力項目は必要に応じて入力しますが、今回は何も入力しません。

次に「ペイロード」タブを開きます。
"Application payload" フィールドには、"Spark エンジン アプリケーション作成 REST API エンドポイントが受け付けるアプリケーション ペイロード JSON" を指定します。
今回は下記の内容を入力します。
重要な点は "ZenApiKey" の値を "spark.hadoop.wxd.apikey" として"conf"の中に指定する事です。 この点がマニュアルでは非常に判りにくい記述となっています(Submitting Spark runtimes by accessing storage の Submitting an application by accessing the watsonx.data catalog に記述されている)。

{
  "application_details": {
    "application": "s3a://spark-bucket0610/spark-sample.py",
    "arguments": [],
    "spark_version": "3.5",
    "conf": {
      "spark.app.name": "pyspark_sample",
      "spark.driver.cores": "1",
      "spark.driver.memory": "4G",
      "spark.executor.cores": "1",
      "spark.executor.memory": "4G",
      "ae.spark.executor.count": "1",
      "spark.hadoop.wxd.apikey": "ZenApiKey Y3BhZG1pbjp.....Uk5Ym1VCg=="
    }
  }
}

「ペイロード」を入力したら右下の「アプリケーションのサブミット」をクリックします。

入力内容に不備がなければ Spark ジョブがサブミットされて実行されます。
Spark エンジンの「アプリケーション」タブに戻り、Spark ジョブのサブミットが受け付けられた事を確認します。

Spark ジョブがエラー無く実行されると「完了」と表示されます。
Spark ジョブ実行中にエラーが発生した場合は「失敗」と表示されます。

Spark アプリケーションの実行結果の確認

Spark ジョブが完了した場合、下記の手順でアプリケーションの処理の結果を確認します。
Spark ジョブが失敗した場合も同じ手順で Spark ジョブが失敗した原因を特定します。

1.Spark エンジンの「アプリケーション」タブで、アプリケーションのIDを確認します。今回は最新の"ea22462c-c072-4d8a-a3"と表示されているアプリケーションを確認します。(アプリケーションIDの一部しか表示されていませんが、通常は先頭の数桁で確認できます)
2.watsonx.data の Webコンソールではなく、CP4Dの Webコンソールを開き右上の ナイン・ドット メニューから「Software Hub」のパースペクティブに切り替えます。

3.ナビゲーション・メニューから「ストレージ・ボリューム」を選択します。

4.「ストレージ・ボリューム」の画面から、ストレージボリュームをクリックします。
(Spark エンジンを追加した時に指定したボリュームが表示されています)

5.「ファイル・ブラウザー」タブをクリックし、左側のペインに表示されている "spark"以下を展開して、ジョブの一覧を表示させます。
ジョブの一覧の中から"ea22462c-c072-4d8a-a3"で始まるジョブを展開し、logs をクリックします。ジョブの一覧の中に、該当するジョブIDが表示されていない場合は、ブラウザをリロードしてみてください。

6.右側のペインに表示されるログの一覧から、ファイル名の最後が"stdout"のログを選択します。

7.「ダウンロード」をクリックします。

8.ダウンロードされたファイルをテキスト・エディターで開きます。
下記のような内容が表示され、最後の行に Python スクリプトで実行された、print の
実行結果が確認できます。実行されたスクリプトの標準出力に書き出された内容は、この stdout のファイルから確認する事ができます。

Warning: Ignoring non-Spark config property: ae.spark.history-server.cores
Warning: Ignoring non-Spark config property: ae.spark.history-server.memory
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
This is a sample spark application

下記の例は失敗した Spark ジョブの stdout の内容ですが、スクリプトの中で文法エラーが発生している事が確認できます。

Warning: Ignoring non-Spark config property: ae.spark.history-server.cores
Warning: Ignoring non-Spark config property: ae.spark.history-server.memory
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
  File "/tmp/spark-b4312b2d-a5c1-4259-b4fb-b00f112a4159/spark-sample.py", line 15
    print("This is sample spark application")
    ^^^^^
SyntaxError: expected 'except' or 'finally' block

スクリプトを修正し、バケットにコピーして再度 Spark ジョブとして実行します。

以上で watsonx.data の Webコンソールから Spark ジョブを手動で実行する手順の例の紹介を終わります。