はじめに
watsonx.data の コマンドライン・インターフェイス による データの取り込み (本文では以下 Ingestion) は、watsonx.data 初期バージョンよりクライアントパッケージ (ibm-lh-client) を使用して実施してきましたが、watsonx.data の次期バージョンである 2.3.0 からクライアントパッケージが廃止になる事が発表されており、替わりに「IBM Cloud Pak for Data コマンドラインインターフェイス (cpdctl)」というCLIを使用して実行する必要があります。
本記事では "cpdctl wx-data" (watsonx.data をサポートする cpdctl コマンド) を使用した Ingestion の実行方法についてご紹介します。
「IBM Cloud Pak for Data コマンドラインインターフェイス (cpdctl)」を初めて使用される方は下記の記事を参照してください。
IBM Cloud Pak for Data 5.2.2 で コマンドライン・インターフェイス cpdctl を実行してみた
参照マニュアル
IBM Cloud Pak for Data コマンドラインインターフェイス (IBM cpdctl)
ingestion
事前準備
環境の準備
本記事の前提として下記の環境を準備しました。
・IBM Cloud Pak for Data 5.2.2 オンプレ環境
・watsonx.data 2.2.2
・Presto (Java)エンジン
・Spark エンジン
・cpdctl を実行するクライアント用インテル系 Linuxマシン
・cpdctl v1.8.45
Ingestionで取り込むデータ(ソース)、取り込み先(ターゲット)の表の準備
Ingestionで取り込むデータ、取り込み先の表を予め準備しておく必要があります。
本検証では取り込むデータは、IBM Cloud Object Storage に作成したバケット内と、クライアント・マシン上に用意しました。
ソース側のバケットとターゲット側のバケットはインフラストラクチャー・マネージャー状で、
取り込み先の表は、IBM Cloud Object Storage に作成したバケット内に用意しました。
取り込み先の表は、watsonx.data のWebコンソールのインフラストラクチャー・マネージャーでApache Iceberg タイプとして登録したターゲット用のカタログにスキーマを作成し、スキーマの下に表を作成しておきます。
事前準備が完了した状態のインフラストラクチャー・マネージャーの表示は以下のようになります。watsonx.data ではバケットはストレージとして表示されます。
(Ingestion 以外の用途のカタログやバケットも表示されています)
Sparkエンジンを使用したIngestionの実行
Parquetファイルの Ingestion 例
指定するパラメーター
| パラメーター | パラメーターの意味 |
|---|---|
| job-id | ジョブID (オプション) |
| engine-id | SparkエンジンのID |
| source-data-files | ソースファイル |
| target-table | ターゲット表 |
| sync-status | 同期モード |
job-id を指定しない場合、cli-ingestion-XXXXX-YYYYYYYYYYYYY の形式のジョブIDが割り当てられます。
engine-id はインフラストラクチャー・マネージャーで SparkエンジンのIDを確認して指定します。
source-data-files は、ソースファイルがバケット内にある場合、"s3://<バケット名>/<ソースファイル名>" の形式で指定します。
target-table は、<カタログ名>.<スキーマ名>.<表名> の形式で指定します。
"sync-status=true" を指定しない場合、IngestionのSparkジョブがサブミットされるとコマンドは終了するため "cpdctl wx-data ingestion get" コマンドでジョブのStatusを確認する必要があります。"sync-status=true" を指定すると実行結果が返るまでコマンドは終了しないため、実行結果を確認する事ができます。
コマンド実行例
jobid=parquet_ingestion1
$ cpdctl wx-data ingestion create \
--engine-id spark843 \
--job-id ${jobid}
--source-data-files s3://src-bucket8/parquet_folder/GVT-DATA-C.txt.v1.parquet \
--target-table tgtbucket8.target8.gvt_data_v \
--sync-status=true
application_id 10e07f66-8954-4e0c-865f-3c13e9680aed
created_at 1764658037097811037
job_id arquet_ingestion1
status starting
...
Ingestion job parquet_ingestion1 status: starting
...
Ingestion job parquet_ingestion1 status: accepted
...
Ingestion job parquet_ingestion1 status: starting
...
Ingestion job parquet_ingestion1 status: running
...
Ingestion job parquet_ingestion1 status: finished
Ingestion job with job ID parquet_ingestion1 has completed successfully.
CSV ファイルの Ingestion 例
指定するパラメーター
| パラメーター | パラメーターの意味 |
|---|---|
| job-id | ジョブID (オプション) |
| engine-id | SparkエンジンのID |
| source-file-type | ソースファイルのタイプ |
| source-data-files | ソースファイル |
| target-table | ターゲット表 |
| file-format-properties | CSVファイルのプロパティー |
| sync-status | 同期モード |
CSVファイルの Ingestion を実行する場合 "source-file-type=csv" を指定し、"file-format-properties" でCSVファイルのプロパティーを下記の形式で指定する必要があります。
{"header": true/false,"encoding":"<value>", "escape_character":"<value>","line_delimiter":"<value>", "field_delimiter":"<value>"}
コマンド実行例
$ filename=staff-dbcs-UTF8-header-dbcs.csv
$ tablename=staff_jp
$ cpdctl wx-data ingestion create \
--engine-id spark843 \
--source-file-type=csv \
--source-data-files s3://src-bucket8/csv_folder/${filename} \
--target-table tgtbucket8.target8.${tablename} \
--file-format-properties '{
"header": true,
"encoding": "utf-8",
"escape_character": "\\",
"line_delimiter": "\n",
"field_delimiter": ","
}' \
--sync-status=true
application_id 20a1c441-a777-4c12-b5b9-2f0d9bde35d9
created_at 1764659081514895970
job_id cli-ingestion-3302f-1764659076971
status starting
...
Ingestion job cli-ingestion-3302f-1764659076971 status: starting
...
Ingestion job cli-ingestion-3302f-1764659076971 status: accepted
...
Ingestion job cli-ingestion-3302f-1764659076971 status: starting
...
Ingestion job cli-ingestion-3302f-1764659076971 status: running
...
Ingestion job cli-ingestion-3302f-1764659076971 status: finished
Ingestion job with job ID cli-ingestion-3302f-1764659076971 has completed successfully.
Lite Ingestion の例
Lite Ingestion はローカル・システムから2MB未満のファイルをIngestする場合に使用できる高速なIngestionです。
指定するパラメーター
| パラメーター | パラメーターの意味 |
|---|---|
| job-id | ジョブID (オプション) |
| engine-id | lite-ingestion |
| source-data-files | ローカル・ソースファイル |
| target-table | ターゲット表 |
| storage-name | 一時領域として使用するバケット名 |
| sync-status | 同期モード |
source-data-files はクライアント・マシン上のソースファイル名を指定します。
実行例
cpdctl wx-data ingestion create \
--engine-id lite-ingestion \
--source-data-files /home/kanbayashi/cpdctl_work/GVT-DATA-C.txt.v1.parquet \
--target-table tgtbucket8.target8.gvt_data_v \
--storage-name tgt-bucket8
...
/home/kanbayashi/cpdctl_work/GVT-DATA-C.txt.v1.parquet file reading in progress...
/home/kanbayashi/cpdctl_work/GVT-DATA-C.txt.v1.parquet file reading completed.
...
created_at 2025-12-02T07:26:24.345831194Z
job_id cli-ingestion-3302f-1764660380107
status completed
以上で"cpdctl wx-data" (watsonx.data をサポートする cpdctl コマンド)を使用した Ingestion の実行方法方の紹介を終わります。