watsonx.data に Confluent Tableflow の Iceberg カタログを登録

Posted at 2026-07-30

この記事では、以下を説明します。

Confluent Tableflow（AWS S3 BYOS）の Iceberg テーブルを watsonx.data から参照するための設定手順
watsonx.data の Presto エンジンで Tableflow テーブルを SQL クエリする方法

この記事は「ETLなしで工場センサーデータをAI分析へ — Confluent Tableflow × watsonx.data × IBM Bob で即時データ分析をやってみよう」の前提設定として必要な手順をまとめたものです。

参考: Querying Confluent Tableflow using Presto engine

watsonx.data の Presto は Vended Credentials に非対応

watsonx.data の Presto エンジンは Confluent の Vended Credentials（Confluent Managed Storage）に対応していません。
TableFlow のストレージは AWS S3（BYOS: Bring Your Own Storage） で設定する必要があります。

0. 前提条件

以下の環境が必要です:

アクティブな Confluent Cloud アカウント
BYOS（AWS S3） で Tableflow が有効化済みの Kafka トピック
Tableflow 用の API キーとシークレット（スコープ: Tableflow）
REST Catalog エンドポイント URL
Kafka クラスター ID
IBM Cloud の watsonx.data インスタンス（Lite 以上）
稼働中の Presto または Prestissimo エンジン
AWS IAM ユーザーの静的な Access Key と Secret Key

尚、watsonx.data の Presto エンジンは IAM ロール（AssumeRole）に非対応のため、IAM ユーザーのアクセスキーが必要です。

1. AWS S3 アクセスキーを取得する

IAM ユーザーの作成

AWS Console で IAM → Users → Create user をクリック
User name を入力（例: watsonxdata-tableflow-user）
Next → Attach policies directly を選択
Tableflow の S3 バケットにアクセスできるポリシーを選択してアタッチ
Next → Create user

アクセスキーの発行

作成したユーザーをクリックして詳細を開く
Security credentials タブを開く
Access keys セクション → Create access key をクリック
Use case で Other を選択して Next
任意の説明タグを入力して Create access key
Access key と Secret access key を必ずコピーして保存する

※ このページを閉じると Secret access key は二度と表示されません。必ずこのタイミングで保存してください。

2. Confluent 側の情報を収集する

REST Catalog エンドポイントの取得

Confluent Cloud Console にログインし、対象の Cluster を開く
左メニュー Tableflow をクリック
API access セクションの REST Catalog Endpoint をコピーする

https://tableflow.<CLOUD_REGION>.aws.confluent.cloud/iceberg/catalog/organizations/<ORG_ID>/environments/<ENV_ID>

Tableflow API キーの作成

TableflowのAPIキーがない場合は作成してください。

API access セクション → Create/View API keys → Add API key
Service account または My account を選択して Next
Resource scope で Tableflow を選択して Next
Create API key をクリックし、API Key と Secret を保存する

Kafka クラスター ID の確認

対象の Cluster → Overview タブを開く
ID（例: lkc-xxxx）をコピーする

※ このクラスター ID は watsonx.data 上でスキーマ名になります。手順 5 の SQL 実行の際に使用します。

3. watsonx.data に Custom データソースを登録する

通常の「Add catalog」ではなく、インフラストラクチャー・マネージャーの「Add data source」→「Custom」 から登録します。

watsonx.data コンソールにログインする
左メニュー インフラストラクチャー・マネージャー を開く
コンポーネントの追加をクリック
データ・ソースのCustom を選択し、「次へ」をクリック
表示名 に任意の名前を入力する（例: confluent_tableflow）
Properties セクションに以下のプロパティを追加する

connector.name=iceberg
iceberg.catalog.type=REST
iceberg.rest.uri=<手順 2 で取得した REST Catalog Endpoint>
iceberg.rest.auth.type=OAUTH2
iceberg.rest.auth.oauth2.credential=<手順 2で取得したTableflow APIKEY>:<手順 2で取得したTableflow APIKEYのSECRET>
hive.s3.aws-access-key=<手順 1 で取得した Access Key>
hive.s3.aws-secret-key=<手順 1 で取得した Secret Key>