さまざまな表形式での作業

適用範囲 ：スパークエンジングルテン加速スパークエンジン

このトピックでは、 Apache Hudi、 Apache Iceberg、 Delta Lake カタログのような異なるテーブル形式にデータを取り込む Spark アプリケーションを実行する手順について説明します。

Spark アプリケーションで使用するデータを格納するために、必要なカタログ（カタログは Apache Hudi, Apache Iceberg または Delta Lake ）を持つストレージを作成する。ストレージを作成するには、ストレージとカタログのペアの追加を参照してください。
ストレージをネイティブ Spark エンジンに関連付けます。詳しくは、エンジンへのカタログの関連付けを参照してください。
Spark アプリケーションを保管するための Cloud Object Storage (COS) を作成します。 Cloud Object Storage およびバケットを作成するには、ストレージ・バケットの作成を参照してください。
Cloud Object Storage を watsonx.dataに登録します。詳しくは、ストレージとカタログのペアの追加を参照してください。
選択したカタログに基づいて、以下の Spark アプリケーション (Python ファイル) をローカル・マシンに保存します。ここ ( iceberg_demo.py、 hudi_demo.py、または delta_demo.py ) で、Spark アプリケーションを COS にアップロードします。データのアップロードを参照してください。

Cloud Object Storageにあるデータを使用して Spark アプリケーションを実行依頼するには、パラメーター値を指定し、以下の表から curl コマンドを実行します。

Apache Iceberg

サンプル・ファイルは、以下の機能を示しています：

watsonx.dataからテーブルにアクセスする
データをwatsonx.dataに取り込む
watsonx.dataのスキーマを変更する

watsonx.dataでテーブルのメンテナンス作業を行う。

データを COS バケットに挿入する必要があります。詳しくは、COSバケットにサンプルデータを挿入するをご覧ください。

Pythonアプリケーション：IcebergPythonファイル

Pythonアプリケーションを送信するCurlコマンド：

サンプル V2 API

curl --request POST \
--url https://<wxd_host_name>/lakehouse/api/v2/spark_engines/<spark_engine_id>/applications \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--header 'LhInstanceId: <instance_id>' \
--data '{  "application_details": {
           "conf": {
                  "spark.hadoop.wxd.apiKey":"Basic <user-authentication-string>"    },
                   "application": "s3a://<application-bucket-name>/iceberg.py"  }
        }'

サンプル V3 API

curl --request POST \
--url https://<wxd_host_name>/lakehouse/api/v3/spark_engines/<spark_engine_id>/applications \
--header 'Authorization: Bearer <token>' \
--header 'Content-Type: application/json' \
--header 'LhInstanceId: <instance_id>' \
--data '{  "application_details": {
           "conf": {
                  "spark.hadoop.wxd.apiKey":"Basic <user-authentication-string>"    },
                   "application": "s3a://<application-bucket-name>/iceberg.py"  }
        }'

パラメータ値：

<wxd_host_name>:watsonx.dataクラウドインスタンスのホスト名。
<instance_id>: watsonx.data インスタンス URL のインスタンス ID。例えば、1609968977179454。
<spark_engine_id>: ネイティブSparkエンジンのエンジンID。
<token>:ベアラートークン。トークンの生成の詳細については、IAM トークンを参照してください。
<user-authentication-string>: ユーザーIDとAPIキーをBase64でエンコードした文字列。フォーマットの詳細については、注釈を参照のこと。
Apache Hudi

PythonSparkアプリケーションは、以下の機能を実証している：

Apache Hudi （データを保存するために作成した）カタログの中にデータベースを作成します。ここでは、'<database_name>.
<database_name> データベース内に'<table_name> というテーブルを作成する。
<table_name> にデータを挿入し、SELECTクエリー操作を行う。
使用後はテーブルとスキーマを削除する。

Pythonアプリケーション：

from pyspark.sql import SparkSession
def init_spark():
    spark = SparkSession.builder.appName("CreateHudiTableInCOS").enableHiveSupport().getOrCreate()
    return spark
def main():
    try:
        spark = init_spark()
        spark.sql("show databases").show()
        spark.sql("create database if not exists spark_catalog.<database_name> LOCATION 's3a://<data_storage_name>/'").show()
        spark.sql("create table if not exists spark_catalog.<database_name>.<table_name> (id bigint, name string, location string) USING HUDI OPTIONS ('primaryKey' 'id', hoodie.write.markers.type= 'direct', hoodie.embed.timeline.server= 'false')").show()
        spark.sql("insert into <database_name>.<table_name> VALUES (1, 'Sam','Kochi'), (2, 'Tom','Bangalore'), (3, 'Bob','Chennai'), (4, 'Alex','Bangalore')").show()
        spark.sql("select * from spark_catalog.<database_name>.<table_name>").show()
        spark.sql("drop table spark_catalog.<database_name>.<table_name>").show()
        spark.sql("drop schema spark_catalog.<database_name> CASCADE").show()
    finally:
        spark.stop()
if __name__ == '__main__':
    main()

パラメーター値:

<database_name>:作成するデータベース名を指定します。
<table_name>:作成するテーブルの名前を指定します。
<data_storage_name>:作成した Apache Hudi ストレージの名前を指定する。

Python アプリケーションをサブミットするための Curl コマンド

サンプル V2 API


curl --request POST
    --url https://<wxd_host_name>/lakehouse/api/v2/spark_engines/<spark_engine_id>/applications
    --header 'Authorization: Bearer <token>'
    --header 'Content-Type: application/json'
    --header 'LhInstanceId: <instance_id>'
    --data '{     "application_details": {
            "conf": {
                    "spark.sql.catalog.spark_catalog.type": "hive",
                    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog",
                    "spark.hadoop.wxd.apiKey":"Basic <user-authentication-string>"        },
                    "application": "s3a://<data_storage_name>/hudi_demo.py"    }}

サンプル V3 API


curl --request POST
    --url https://<wxd_host_name>/lakehouse/api/v3/spark_engines/<spark_engine_id>/applications
    --header 'Authorization: Bearer <token>'
    --header 'Content-Type: application/json'
    --header 'LhInstanceId: <instance_id>'
    --data '{     "application_details": {
            "conf": {
                    "spark.sql.catalog.spark_catalog.type": "hive",
                    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.hudi.catalog.HoodieCatalog",
                    "spark.hadoop.wxd.apiKey":"Basic <user-authentication-string>"        },
                    "application": "s3a://<data_storage_name>/hudi_demo.py"    }}

パラメーター値:

<wxd_host_name>:watsonx.dataクラウドインスタンスのホスト名。
<instance_id>: watsonx.data インスタンス URL のインスタンス ID。例えば、1609968977179454。
<spark_engine_id>: ネイティブSparkエンジンのエンジンID。
<token>:ベアラートークン。トークンの生成の詳細については、IAM トークンを参照してください。
<user-authentication-string>: ユーザーIDとAPIキーをBase64でエンコードした文字列。フォーマットの詳細については、注釈を参照のこと。
Delta Lake

PythonSparkアプリケーションは、以下の機能を実証している：

Delta Lake （データを保存するために作成した）カタログの中にデータベースを作成します。ここでは、'<database_name>.
<database_name> データベース内に'<table_name> というテーブルを作成する。
<table_name> にデータを挿入し、SELECTクエリー操作を行う。
使用後はテーブルとスキーマを削除する。

Pythonアプリケーション：

from pyspark.sql import SparkSession
import os
    def init_spark():
         spark = SparkSession.builder.appName("lh-hms-cloud").enableHiveSupport().getOrCreate()
         return spark
    def main():
             spark = init_spark()
             spark.sql("show databases").show()
                     spark.sql("create database if not exists spark_catalog.<database_name> LOCATION 's3a://<data_storage_name>/'").show()
                     spark.sql("create table if not exists spark_catalog.<database_name>.<table_name> (id bigint, name string, location string) USING DELTA").show()
                     spark.sql("insert into spark_catalog.<database_name>.<table_name> VALUES (1, 'Sam','Kochi'), (2, 'Tom','Bangalore'), (3, 'Bob','Chennai'), (4, 'Alex','Bangalore')").show()
                     spark.sql("select * from spark_catalog.<database_name>.<table_name>").show()
                     spark.sql("drop table spark_catalog.<database_name>.<table_name>").show()
                     spark.sql("drop schema spark_catalog.<database_name> CASCADE").show()
                     spark.stop()
    if __name__ == '__main__':
        main()

パラメーター値:

<database_name>:作成するデータベース名を指定します。
<table_name>:作成するテーブルの名前を指定します。
<data_storage_name>:作成した Apache Hudi ストレージの名前を指定する。

Python アプリケーションをサブミットするための Curl コマンド

サンプル V2 API

curl --request POST
--url https://<wxd_host_name>/lakehouse/api/v2/spark_engines/<spark_engine_id>/applications
--header 'Authorization: Bearer <token>'
--header 'Content-Type: application/json'
--header 'LhInstanceId: <instance_id>'
--data '{        "application_details": {
       "conf": {
       "spark.sql.catalog.spark_catalog" : "org.apache.spark.sql.delta.catalog.DeltaCatalog",
       "spark.sql.catalog.spark_catalog.type" : "hive",
       "spark.hadoop.wxd.apiKey":"<user-authentication-string>"        },
       "application": "s3a://<database_name>/delta_demo.py"        }    }

サンプル V3 API

curl --request POST
--url https://<wxd_host_name>/lakehouse/api/v3/spark_engines/<spark_engine_id>/applications
--header 'Authorization: Bearer <token>'
--header 'Content-Type: application/json'
--header 'LhInstanceId: <instance_id>'
--data '{        "application_details": {
       "conf": {
       "spark.sql.catalog.spark_catalog" : "org.apache.spark.sql.delta.catalog.DeltaCatalog",
       "spark.sql.catalog.spark_catalog.type" : "hive",
       "spark.hadoop.wxd.apiKey":"<user-authentication-string>"        },
       "application": "s3a://<database_name>/delta_demo.py"        }    }

パラメーター値

<wxd_host_name>:watsonx.dataクラウドインスタンスのホスト名。
<instance_id> watsonx.data クラスタ・インスタンス URL のインスタンス ID。例えば、1609968977179454。
<spark_engine_id>: ネイティブSparkエンジンのエンジンID。
<token> ベアラートークントークンの生成の詳細については、IAM トークンを参照してください。
<user-authentication-string>: ユーザーIDとAPIキーをBase64でエンコードした文字列。フォーマットについて詳しくは、以下の注を参照してください。

<user-authentication-string> の値は、 echo -n 'ibmlhapikey_<username>:<user_apikey>' | base64 の形式でなければなりません。ここで、 <user_id> は、データ・バケットへのアクセスに使用される API キーを持つユーザーの IBM Cloud ID です。ここで、 <IAM_APIKEY> は、オブジェクト・ストア・バケットにアクセスするユーザーの API キーです。 API キーを生成するには、 watsonx.data コンソールにログインし、「プロファイル」>「プロファイルと設定」>「API キー」にナビゲートして、新しい API キーを生成します。新しい API キーを生成すると、古い API キーが無効になります。

Spark アプリケーションをサブミットすると、アプリケーション ID と Spark バージョンを示す確認メッセージが表示されます。参照用に保存します。
watsonx.data クラスターにログインし、「エンジンの詳細」ページにアクセスします。「アプリケーション」タブで、アプリケーション ID を使用してアプリケーションをリストし、ステージを追跡します。詳しくは、アプリケーションの表示および管理を参照してください。

さまざまな表形式での作業

さまざまな表形式での作業

関連API