ETLなしで工場センサーデータをAI分析へ — Confluent Tableflow × watsonx.data × IBM Bob で即時データ分析をやってみよう

Last updated at 2026-07-31Posted at 2026-07-30

この記事で使用するコードはすべて GitHub リポジトリ confluent-tableflow-wxd に置いています。
コードはIBM Bobにて作成しました。

はじめに

「Kafka にリアルタイムデータを流しているのに、分析は翌日バッチ…」という状況はよくありますね。

この記事では Confluent Tableflow を使って、Kafka Topic のデータを ETL なしで Apache Iceberg テーブルへ自動変換し、IBM watsonx.data の Presto エンジンと IBM Bob（AI Agent）で即時に故障リスクを判定してみます。

コードは最小限、クリック操作がメインです。製造業の設備センサーデータを題材にしていますが、IoT データであればどの業種にも応用できます。

当記事でやること

Confluent Tableflow で Kafka → Iceberg テーブルを 数クリックで有効化する
watsonx.data の Query Editor で Iceberg テーブルに SQL で直接クエリする
IBM Bob に自然言語で問いかけ、故障リスクを判定させる
データを追加するだけで AI の判定が変わる様子を確認する

アーキテクチャ

前提条件

以下のアカウント・環境が必要です。

必要なもの	備考
Confluent Cloud上のKafkaクラスター	Confluent Cloud
watsonx.data SaaSprestoエンジン	Lite プランで試せます
IBM Bob（watsonx.data MCP 接続済み）	MCP 設定はこちらの記事を参照してください
Python 3.13 以上 + uv	`brew install uv` など。pythonに詳しい方ならuvなくてもpython実行環境があればOK(本記事はuv前提の記載)
AWS confluent clusterと同一リーションにあるS3バケット	ICOS不可
AWS CLI	デモリセット時に使用

事前に完了させておく設定

以下 2 点は、別の手順が必要なため先に済ませておいてください。

S3 の Provider integration 設定
Confluent Cloud コンソールで S3 バケットへのアクセスを許可する設定です。
→ Configure Storage for Tableflow in Confluent Cloud: Bring Your Own Storage (BYOS) →Amazon S3
watsonx.data への Iceberg カタログ登録
Tableflow が書き出す S3 パスを watsonx.data のカタログとして登録する手順です。
→ watsonx.data に Confluent Tableflow の Iceberg カタログを登録（別記事）

事前準備

1. コードのセットアップ

GitHub からリポジトリをクローンし、依存ライブラリをインストールします。

git clone https://github.com/kyokonishito/confluent-tableflow-wxd.git
cd confluent-tableflow-wxd
uv sync

以降のコマンドはすべて confluent-tableflow-wxd ディレクトリで実行してください。

2. Confluent Cloud — Topic 作成

Confluent Cloud コンソールで以下の Topic を作成します。

項目	値
Topic 名	`factory_machine_events`
Partitions	1
Retention	7 days（デフォルト）

Topic 名はデモスクリプトの .env に合わせています。変更する場合は後述の .env も更新してください。

3. Schema Registry — Avro スキーマ登録

Confluent Cloud コンソールの Schema Registry タブで、Subject factory_machine_events-value に以下の Avro スキーマを登録します。

{
  "type": "record",
  "name": "FactoryMachineEvent",
  "namespace": "com.demo.factory",
  "fields": [
    { "name": "timestamp",   "type": "string" },
    { "name": "machine_id",  "type": "string" },
    { "name": "temperature", "type": "double" },
    { "name": "vibration",   "type": "double" },
    { "name": "power",       "type": "double" }
  ]
}

Tableflow は Schema Registry のスキーマを元に Iceberg テーブルのカラム定義を自動生成します。スキーマ登録は必ず先に行ってください。

4. `.env` ファイルを作成

.env.example をコピーして .env を作成し、Confluent の接続情報を入力します。

cp .env.example .env

.env に以下の値を設定してください。

# ================================================================== #
# Confluent Cloud 接続設定
# ================================================================== #
CONFLUENT_BOOTSTRAP_SERVERS=pkc-xxxxx.us-east-1.aws.confluent.cloud:9092
CONFLUENT_API_KEY=your_api_key
CONFLUENT_API_SECRET=your_api_secret
CONFLUENT_TOPIC=factory_machine_events
## （デモリセット時に使用）
CONFLUENT_CLUSTER_ID=lkc-xxxxxxx

# Schema Registry 接続設定
CONFLUENT_SCHEMA_REGISTRY_URL=https://psrc-xxxxx.us-east-1.aws.confluent.cloud
CONFLUENT_SCHEMA_REGISTRY_API_KEY=your_sr_api_key
CONFLUENT_SCHEMA_REGISTRY_API_SECRET=your_sr_api_secret

# watsonx.data 接続設定（デモリセット時に使用）
WXD_URL=https://us-south.lakehouse.cloud.ibm.com
WXD_API_KEY=your_ibmcloud_api_key
WXD_INSTANCE_CRN=crn:v1:bluemix:public:lakehouse:us-south:a/your_account_id:your_instance_uuid::
WXD_ENGINE_ID=your_presto_engine_id

接続情報の確認場所：

CONFLUENT_API_KEY / CONFLUENT_API_SECRET：Confluent Cloud API Key の作成方法を参照してください（Kafka cluster スコープ）
CONFLUENT_SCHEMA_REGISTRY_API_KEY / CONFLUENT_SCHEMA_REGISTRY_API_SECRET：Confluent Cloud API Key の作成方法を参照してください（Schema Registry スコープ）
CONFLUENT_BOOTSTRAP_SERVERS / CONFLUENT_CLUSTER_ID：Confluent Cloud コンソールの Bootstrap server, ID から取得します。
CONFLUENT_SCHEMA_REGISTRY_URL：Confluent Cloud コンソールの Endpoints から取得します。
WXD_*：watsonx.data コンソールから取得します

Step 1. Tableflow を有効化する

事前準備が完了したら、Tableflow を有効化します。この作業は数クリックで完了します。

Confluent Cloud コンソールで Topic factory_machine_events を開く
「Enable Table Flow」 をクリック
「Iceberg」にチェックを入れ、左の 「Configure Custom Storage」 をクリック
「Store in your own storage」にチェックを入れ、Provider integration と AWS S3 Bucket name を指定して 「Continue」 をクリック
「Launch」 をクリックして有効化

数分後、watsonx.data のカタログにテーブルが出現します。

Step 2. 正常センサーデータを投入する

コマンドはすべて confluent-tableflow-wxd ディレクトリで実行してください。

正常稼働中のセンサーデータ（PRESS-001・PRESS-002 各 30 件）を Kafka に投入します。

uv run produce.py demo_data_normal.json --interval 0.1 --segment-flush

実行すると以下のように出力されます。

producer生成: 0.8s
file: demo_data_normal.json  topic: factory_machine_events  records: 60件  interval: 0.1s  loop: False

  → 2026-07-25T10:00:00  temp=65.2  vib=0.12  power=220.0
  → 2026-07-25T10:00:00  temp=65.2  vib=0.12  power=220.0
  ...（60件）

完了

[segment-flush] ダミーレコード 1,800,000 件（約100MB）を一括送信してセグメントを強制ロール...
[segment-flush] 完了。Tableflow が約1〜2分以内に Iceberg へ反映します。

オプション --segment-flush

Tableflow は Kafka のセグメントがロールされたタイミングで Iceberg snapshot をコミットします。通常は最大 15 分かかりますが、--segment-flush を付けると約 100MB のダミーレコードを送信してセグメントを強制ロールするため、約 1〜2 分 で反映されます。

送信自体は約 15 秒で完了します。ダミーレコードの machine_id は FLUSH-000 なので、SQL クエリで WHERE machine_id != 'FLUSH-000' と除外できます。

実運用ではこのダミー投入は不要です。 最大 15 分で自動的に反映されます。

Step 3. watsonx.data で SQL 確認（正常状態）

1〜2 分待ったら、watsonx.data の Query Editor でデータを確認しましょう。

以降の SQL の "<カタログ名>"."<Cluster ID>"."factory_machine_events" は、お使いの環境のカタログ名・スキーマ名に合わせて変更してください。
カタログ名・スキーマ名は watsonx.data コンソール > データ・マネージャー で確認できます。
スキーマ名はCluster IDです。

件数確認

SELECT machine_id, count(*) AS cnt
FROM "<カタログ名>"."<Cluster ID>"."factory_machine_events"
WHERE machine_id != 'FLUSH-000'
GROUP BY machine_id;

期待結果：

machine_id	cnt
PRESS-001	30
PRESS-002	30

平均値確認

SELECT
    machine_id,
    round(avg(temperature), 1) AS avg_temp,
    round(avg(vibration), 3)   AS avg_vibration,
    max(temperature)           AS max_temp,
    max(vibration)             AS max_vibration
FROM "<カタログ名>"."<Cluster ID>"."factory_machine_events"
WHERE machine_id != 'FLUSH-000'
GROUP BY machine_id;

期待結果：

machine_id	avg_temp	avg_vibration	max_temp	max_vibration
PRESS-001	65.2	0.120	65.2	0.120
PRESS-002	65.2	0.120	65.2	0.120

2 台とも同じ正常値で安定していることが確認できます。

Step 4. IBM Bob で異常検知① — 正常データのみ

watsonx.data に MCP 接続した IBM Bob に、自然言語で故障リスク判定を依頼します。

IBM Bob の MCP 設定がまだの方は、こちらの記事を参考に設定してください。

IBM Bob へのプロンプト

IBM Bob を Agent モードで開き、以下を入力します（テーブル名はご自身の環境に合わせてください）。

"<カタログ名>"."<Cluster ID>"."factory_machine_events" には PRESS-001・PRESS-002 の2台分の工場設備センサーデータが入っています。FLUSH-000 はダミーデータなので除外してください。
各設備の異常なレコードを特定し、リスクレベルを判定してください。異常データがなければ正常と報告お願いします。

IBM Bob の回答例（正常データのみ）

(詳細のの内容がある場合はHTMLファイルで作成してくれます。)

PRESS-001：正常（リスクレベル：なし）
温度・振動・電力のすべてが安定した範囲内で推移。外れ値・急変なし。

PRESS-002：正常（リスクレベル：なし）
PRESS-001 と同等の安定した値域を維持。外れ値・急変なし。

センサー	最小値	最大値	平均値
温度	65.1 °C	65.2 °C	65.2 °C
振動	0.12	0.12	0.12
電力	219.9 W	220.1 W	220.0 W

正常時のベースラインが記録されました。次のステップで「データだけを変えると AI の判断も変わる」ことを確認します。

Step 5. 異常データを投入してリアルタイム検知

同じ質問・同じ SQL・同じプロンプトのまま、データだけを追加します。

異常データを投入

demo_data_anomaly.json には、PRESS-001 の異常値 5 件と PRESS-002 の正常値 5 件が入っています。

uv run produce.py demo_data_anomaly.json --interval 0.1 --segment-flush

1〜2 分待ったら watsonx.data で件数を再確認します。

SELECT machine_id, count(*) AS cnt
FROM "<カタログ名>"."<Cluster ID>"."factory_machine_events"
WHERE machine_id != 'FLUSH-000'
GROUP BY machine_id;

期待結果：

machine_id	cnt
PRESS-001	35
PRESS-002	35

各 30 件 → 35 件に増えていれば、Tableflow が Iceberg へ反映できています。

IBM Bob に同じプロンプトで再質問

Step 4 とまったく同じプロンプトをもう一度送ります。

"<カタログ名>"."<Cluster ID>"."factory_machine_events" には PRESS-001・PRESS-002 の2台分の工場設備センサーデータが入っています。FLUSH-000 はダミーデータなので除外してください。
各設備の異常なレコードを特定し、リスクレベルを判定してください。異常データがなければ正常と報告お願いします。

IBM Bob の回答例（PRESS-001 に異常データ追加後）

詳細の内容があったのでHTMLファイルで作成してくれました。（イメージは概要の後に添付)
以下は概要です。

PRESS-001 → CRITICAL（即時点検要）
約 1 秒間に 5 レコードが連続記録され、3 センサーがすべて閾値超過。正常時から急激に乖離しています。

レコード	温度	振動	電力	リスク
10:30	85°C	0.42	290 kW	HIGH
10:31	95°C	0.61	320 kW	HIGH
10:32	103°C	0.78	345 kW	CRITICAL
10:33	108°C	0.95	370 kW	CRITICAL
10:34	115°C	1.15	400 kW	CRITICAL

PRESS-002 → NORMAL（対応不要）
全 35 件が正常範囲内（温度 max 65.2°C / 振動 max 0.12 / 電力 max 220.0 W）。

HTMLでの詳細レポート

データが変われば AI の判断も変わります。

Step 6. 自然言語での問い合わせ

保全エンジニア向けの詳細判定に加えて、もっとシンプルな自然言語でも問い合わせられます。

IBM Bob へのプロンプト

今日最も故障リスクが高い設備は？

IBM Bob の回答例

(詳細のの内容がある場合はHTMLファイルで作成してくれます。)

PRESS-001 が本日最高リスク設備です。

3 センサーが同時に急上昇しており、データ末尾時点でも上昇が継続中です。

指標	正常ベースライン	最大値	逸脱率
温度	65°C	115°C	+77%
振動	0.12 mm/s	1.15 mm/s	+858%
電力	220 W	400 W	+82%

PRESS-002 は全センサーが安定しており問題なし。

即時対応として、PRESS-001 の稼働停止と冷却系・潤滑系の点検を推奨します。

Step 7. デモ環境のリセット（オプション）

デモを繰り返す場合は以下の手順で環境を初期化します。

1. S3 パスの特定

Confluent Cloud コンソールで確認します。

Topic factory_machine_events を開く
Overview タブを選択
Tableflow details セクションの Location に表示されている S3 パスをコピーし、どこかに保存する

2. Tableflow を停止 → Topic 削除 → S3 削除 → Topic 再作成

必ず Tableflow を停止してから S3 を削除してください。停止前に削除すると Tableflow が破損する可能性があります。

Confluent コンソールで Topic factory_machine_events を開く
Settings タブ > Tableflow セクション > Disable Tableflow
Topic factory_machine_events を削除
AWS CLI で Iceberg テーブルのデータディレクトリを削除

--profile <profile> には、対象の S3 バケットへアクセスできる AWS CLI プロファイル名を指定します。default プロファイルを使う場合は --profile を省略できます。登録済みプロファイル一覧は aws configure list-profiles で確認できます。
```
export S3_PATH=s3://<bucket>/<uuid-of-factory_machine_events>
aws s3 ls "$S3_PATH" --recursive --profile <profile>
aws s3 rm "$S3_PATH" --recursive --profile <profile>
```
Topic factory_machine_events を同名・同設定で再作成（事前準備 Step 1 参照）
必要なら Subject factory_machine_events-value が残っていることを確認する（事前準備 Step 2 参照）

2. watsonx.data Presto エンジンの再起動

S3 データを削除した後、Presto のメタデータキャッシュをクリアします。

.env の WXD_* 設定を確認してから実行します。

uv run demo_setup.py

=== デモセットアップ開始 ===

[1/4] Kafka トピック segment.ms を 600000ms（10分）に設定...
  完了

[2/4] IBM Cloud IAM トークンを取得...
  完了

[3/4] watsonx.data エンジン (presto741) を再起動...
  再起動リクエスト送信完了。running になるまで待機...

[4/4] エンジン起動待機...
  エンジンステータス: restarting  (0s 経過)
  エンジンステータス: running  (30s 経過)

=== セットアップ完了。デモを開始できます。===

約 30 秒でエンジンが RUNNING に戻り、新しい Iceberg テーブルが正しく参照できる状態になります。

再起動しないと、古いメタデータがキャッシュに残り、テーブルが見えない・件数が 0 件になるなどの問題が発生する可能性があります。

まとめ

当記事では以下をやってみました。

ステップ	内容
Tableflow 有効化	Kafka Topic → Iceberg テーブルを数クリックで構成。ETL・Connector コードなし
センサーデータ投入	`produce.py` で Avro 形式のデータを Kafka へ送信
SQL 確認	watsonx.data の Query Editor で Iceberg テーブルへ直接クエリ
IBM Bob による分析	自然言語プロンプトのみで故障リスク判定。
即時検知	データを追加するだけで AI の判定が変わることを確認

Tableflow によって「Kafka に流れているデータがそのまま分析テーブルになる」という体験が、ETL パイプラインの設計・開発・運用のコストを大幅に削減します。さらに IBM Bob を組み合わせることで、「データが変われば AI の判断も変わる」オンデマンドな分析が実現します。ぜひやってみてください！

ETLなしで工場センサーデータをAI分析へ — Confluent Tableflow × watsonx.data × IBM Bob で即時データ分析をやってみよう

はじめに

当記事でやること

アーキテクチャ

前提条件

事前に完了させておく設定

事前準備

1. コードのセットアップ

2. Confluent Cloud — Topic 作成

3. Schema Registry — Avro スキーマ登録

4. .env ファイルを作成

Step 1. Tableflow を有効化する

Step 2. 正常センサーデータを投入する

Step 3. watsonx.data で SQL 確認（正常状態）

件数確認

平均値確認

Step 4. IBM Bob で異常検知① — 正常データのみ

IBM Bob へのプロンプト

IBM Bob の回答例（正常データのみ）

Step 5. 異常データを投入してリアルタイム検知

異常データを投入

IBM Bob に同じプロンプトで再質問

IBM Bob の回答例（PRESS-001 に異常データ追加後）

HTMLでの詳細レポート

Step 6. 自然言語での問い合わせ

IBM Bob へのプロンプト

IBM Bob の回答例

Step 7. デモ環境のリセット（オプション）

1. S3 パスの特定

2. Tableflow を停止 → Topic 削除 → S3 削除 → Topic 再作成

2. watsonx.data Presto エンジンの再起動

まとめ

関連リンク

4. `.env` ファイルを作成