オンプレの watsonx.data Premium Edition 2.4.x (Software Hub 5.4.x) を導入するとSparkも導入されます。そのSparkを使ってwatsonx.dataに構築されたIcebergテーブルにアクセスするための方法として、以下の4つが挙げられます。
- Watson StudioのJupyterノートブックで対話的に実行
- PCのVisual Studio Code(VSCode)を使ってSpark Labで対話的に実行
- GUIを使ってあらかじめ保存されたPySparkプログラムを実行
- REST APIであらかじめ保存されたPySparkプログラムを実行
この記事ではこれらの使い方のポイントを説明します。
はじめに
watsonx.dataでSparkを使用するためにはwatsonx.dataのインフラストラクチャー・マネージャーからSparkを追加します。追加時に、Sparkで実行したジョブ実行結果の出力先となるファイル・ストレージも定義します。
SparkからアクセスするIcebergのカタログの関連付けも行います。
1. Watson StudioのJupyterノートブックで対話的に実行
最初に、プロジェクトの 管理 ＞ 環境 ＞ テンプレート にwatsonx.dataのSparkを追加します。
Jupyterノートブック・エディタを開く際に、追加した環境を選択します。
Jupyterノートブックでは編集モードにした時点でSparkが自動で起動されますが、Icebergテーブルにアクセスするための認証情報をセットするため、以下の例のように、一旦止めて、username/passwordとZenApiKeyをセットして起動し直します。
以下の代わりに、テンプレートの Spark構成(オプション) に key=value 形式で以下のプロパティ値を設定しておくことも可能です(値は二重引用符等で囲まず、半角スペースで値が区切られる場合もそのまま記載します)。
conf=spark.sparkContext.getConf()
spark.stop()
from pyspark.sql import SparkSession
conf.setAll([ \
("spark.hive.metastore.client.plain.username", "username"), \
("spark.hive.metastore.client.plain.password", "password"), \
("spark.hadoop.wxd.apikey", "ZenApiKey xxx")
])
spark=SparkSession.builder.config(conf=conf).getOrCreate()
実行させるSQL文によって、username/passwordかZenApiKeyで認証されます。
ZenApiKeyの値は以下の値になります。APIキーはSoftware HubまたはIBM Cloud Pak for Data画面の右上のユーザー・アイコンから プロファイルと設定 ＞ API鍵 をクリックして取得します。
echo -n "<ユーザー名>:<APIキー>" | base64
Jupyterノートブックを使用したプログラムの実行例を以下に示します。
2. PCのVisual Studio Code(VSCode)を使ってSpark Labで対話的に実行
watsonx.dataのインフラストラクチャー・マネージャーから得られるSparkの詳細情報に記載されているVSCodeの接続構成(json形式)を使って、PCのVSCodeからPySparkを実行してwatsonx.dataのIcebergテーブルにアクセスすることができます。
VSCodeには、watsonx.dataのextensionを導入します。
watsonx.dataのextensionから Manage Connection をクリックして、API KeyとVSCodeの接続構成情報(Connection JSON)を入力し、Test&Save をクリックします。
該当エンジンの上にマウスを合わせたら表示される + をクリックし、Spark Labの名前やdriver/executorの情報等を入力後、Create をクリックしSparkを起動させます。
Sparkが起動すると該当エンジンの下に作成したSpark Labの名前がリストされます。それをクリックするとSpark Labのワークスペース環境が表示されます。
Spark Labの名前の右に表示されるゴミ箱をクリックすると、Sparkのdriverやexecutorが削除されます。その際、そのSpark Lab上で作成したファイルも削除されますので、必要なファイルは事前にバックアップするなどの対応が必要となります。
Spark Labのワークスペース環境の左側に表示されるExplorerにJupyterノートブック・ファイルをドラッグ&ドロップでインポートしたり、新規にファイルを作成したりして、プログラムを開発&テストしていくことができます。
最初に実行する際、PythonとJupyterのextensionを導入するようにガイドされますので、指示に従って導入します。
以下の例ではPython 3.11.15を使って実行しています。
関連するQiita記事として以下があります。
watsonx.data: Spark lab を使ってVSCodeからSparkアプリを実行する
3. GUIを使ってあらかじめ保存されたPySparkプログラムを実行
3章と4章では以下のプログラムを使用します。username/passwordやZenApiKeyに正しい値をセットし、spark.sqlメソッドで指定しているSQL文も環境に合わせて変更します。
ZenApiKeyの値は 1. Watson StudioのJupyterノートブック・エディタで対話的に実行 で述べた値と同じです。
from pyspark.sql import SparkSession
import sys
def init_spark():
spark = SparkSession.builder\
.config("spark.hive.metastore.client.plain.username", "username") \
.config("spark.hive.metastore.client.plain.password", "password") \
.config("spark.hadoop.wxd.apikey", "ZenApiKey xxx") \
.enableHiveSupport()\
.getOrCreate()
return spark
def list_schema(spark):
spark.sql("show schemas from iceberg_data1").show()
def select_table(spark):
spark.sql("select * from iceberg_data1.test.cars limit 10").show()
def main():
try:
spark = init_spark()
list_schema(spark)
select_table(spark)
finally:
spark.stop()
if __name__ == '__main__':
exit_code = 0
try:
main()
except Exception as e:
print(e)
exit_code = 1
sys.exit(exit_code)
プログラムを実行する前に、IBM Cloud Pak for Dataの 管理 ＞ ストレージ・ボリューム でプログラムを配置するストレージを定義します。
適当なフォルダーを作成してプログラムファイルをアップロードします。以下の例では、app01 フォルダーを作成して、spark_test1.py ファイルをアップロードしています。
PySparkプログラムの実行は、インフラストラクチャー・マネージャー で該当Sparkをクリックし、アプリケーション タブから アプリケーションの作成 をクリックして、各種情報をセットして アプリケーションのサブミット ボタンをクリックして実行します。
上記でアップロードした spark_test1.py プログラムを実行する場合、以下のような値を設定します。
設定値があまり直感的でないものもあり、注意が必要です。
| 項目 | 値 |
|---|---|
| アプリケーション・パス | /app/spark_test1.py |
| ボリューム | cpd::app |
| マウント・パス | /app |
| ソース・サブ・パス | app01 |
以下に上記の値をセットしている設定画面例を示します。画面取得の都合上、ハードウェア構成 フィールドを閉じていますが、ここで、Sparkのdriver/executorに関する情報(CPU/メモリ等)を入力することができます。
アプリケーションをサブミットすると、以下のように、アプリケーションに ID が付与され、その状況がリストされます。
実行結果は、Spark構成時に定義したストレージに保存されます。
(IBM Cloud Pak for Dataの左上のハンバーガーメニュー ＞ 管理 ＞ ストレージ・ボリューム)
最後にリストされている stdout ファイルに、PySparkプログラムの実行結果が保存されています。
WARNING: Using incubator modules: jdk.incubator.vector
+---------+
|namespace|
+---------+
| test|
+---------+
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.
[Stage 0:> (0 + 1) / 1]
+--------------------+----+---------+------------+----------+------+------------+-----+------+
| Car| MPG|Cylinders|Displacement|Horsepower|Weight|Acceleration|Model|Origin|
+--------------------+----+---------+------------+----------+------+------------+-----+------+
|Chevrolet Chevell...|18.0| 8| 307.0| 130.0| 3504| 12.0| 70| US|
| Buick Skylark 320|15.0| 8| 350.0| 165.0| 3693| 11.5| 70| US|
| Plymouth Satellite|18.0| 8| 318.0| 150.0| 3436| 11.0| 70| US|
| AMC Rebel SST|16.0| 8| 304.0| 150.0| 3433| 12.0| 70| US|
| Ford Torino|17.0| 8| 302.0| 140.0| 3449| 10.5| 70| US|
| Ford Galaxie 500|15.0| 8| 429.0| 198.0| 4341| 10.0| 70| US|
| Chevrolet Impala|14.0| 8| 454.0| 220.0| 4354| 9.0| 70| US|
| Plymouth Fury iii|14.0| 8| 440.0| 215.0| 4312| 8.5| 70| US|
| Pontiac Catalina|14.0| 8| 455.0| 225.0| 4425| 10.0| 70| US|
| AMC Ambassador DPL|15.0| 8| 390.0| 190.0| 3850| 8.5| 70| US|
+--------------------+----+---------+------------+----------+------+------------+-----+------+
この例では、プログラムをストレージ・ボリュームに配置して実行していますが、オブジェクト・ストレージに配置して実行することも可能で、以下のQiita記事にそのやり方が記載されています。
watsonx.data の WebコンソールからSpark job を実行してみた
4. REST APIであらかじめ保存されたPySparkプログラムを実行
REST APIの書式は以下のようになります。
curl -k -X POST \
-H "Content-Type: application/json" \
-H "AuthInstanceId: <instanceId>" \
-H "Authorization: ZenApiKey <ZenApiKey>" -d @input.json \
"https://<cpd_host_name>/lakehouse/api/v3/<instanceId>/spark_engines/<spark_engine_id>/applications"
各項目の値は以下になります。
| 項目 | 値 |
|---|---|
| cpd_host_name | Software Hub, IBM Cloud Pak for Data, watsonx.data等のURLに記載されているホスト名 |
| instanceId | インフラストラクチャー・マネージャー等のURLにセットされているinstanceIdパラメーター。IBM Cloud Pak for Data画面のサービス・インスタンスで表示されるlakehouseの詳細画面にも記載されています |
| spark_engine_id | インフラストラクチャー・マネージャーで得られる該当Sparkの詳細画面に記載されているエンジンID |
| ZenApiKey | echo -n "<ユーザー名>:<APIキー>" | base64 |
input.json には、プログラムの情報を記載します。
書式は、API syntax for submitting Spark runtime の Table 1. Parameters for the Spark runtimes API に記載されています。
この例では以下をセットしています。
{
"application_details": {
"application": "/app/spark_test1.py",
"arguments": [],
"conf": {
"spark.app.name": "spark_test1_rest_api",
"spark.driver.memory": "4G",
"spark.driver.cores": 1,
"spark.executor.memory": "4G",
"spark.executor.cores": 1,
"ae.spark.executor.count": 2
}
},
"volumes": [
{
"name": "cpd::app",
"mount_path": "/app",
"source_sub_path": "app01"
}
]
}
実行したコマンドと戻り値を以下に示します。
(ZenApiKeyの値等、一部の値は変更して記載しています)
$ curl -k -X POST \
-H "Content-Type: application/json" \
-H "AuthInstanceId: 1787655754349462" \
-H "Authorization: ZenApiKey xxx" -d @input.json \
"https://cpd-cpd.apps.xxx.com/lakehouse/api/v3/1787655754349462/spark_engines/spark694/applications"
{"id":"f558662a-6e2e-4f1e-8f37-27ff87c7cd4e","state":"accepted","spark_ui":"","template_id":"spark-4.0-cp4d-wxd-template","spark_version":"4.0","runtime":{"spark_version":"4.0"}}
実行したコマンドは、処理が受け付けられた時点でJSON形式の値が戻されます。GUIを使ってアプリケーションをサブミットした時と同様に ID が付与されます。Sparkの実行状況や結果は、3. GUIを使ってあらかじめ保存されたPySparkプログラムを実行 と同様なやり方で確認できます。
