シナリオ: オブジェクト・ストレージからのバッチ・データの取込み

OCI Object Storageから順次CSVオブジェクトをダウンロードし、Node-REDの各バッチ・レコードを変換してOCI IoTに取り込みます。

このシナリオを使用して、オブジェクト・ストレージから順次CSVオブジェクトをダウンロードし、オフラインまたは履歴の読取りをOCI IoTにリプレイして、デジタル・ツイン履歴を完全な状態に保ちます。Node-REDフローはInject -> Prepare Filename -> Object Storage Download -> Increment Index -> CSV -> Transform -> MQTT-OUTです。

ノート

このシナリオでは、各CSVファイルはフローがオンデマンドでダウンロードおよび処理する個別のオブジェクトであるため、オブジェクト・ストレージを使用します。フローで、マウントされたファイル・システム・パスを介して使用可能な共有ファイルが必要な場合は、File Storageを使用します。ファイル・ストレージまたはオブジェクト・ストレージの選択を参照してください。

開始する前に、一般的なシナリオ設定を完了してください。

前提条件

バッチCSVオブジェクトのオブジェクト・ストレージ・バケットを作成します。オブジェクト・ストレージ・ネームスペースおよびバケット名を記録します。フロー・ランタイム動的グループを作成し、オブジェクト・ストレージ・ノードで使用されるリソース・プリンシパル構成に、バケットからオブジェクトをダウンロードするために必要なオブジェクト・ストレージ読取りポリシーがあることを確認します。

フロー・ランタイムは、オブジェクト・ストレージおよびOCI IoTデバイス・ホストに到達できる必要があります。

ステップ1: フロー・ランタイムの作成

OCI Consoleで、ナビゲーション・メニューを開きます。「開発者サービス」に移動し、「Internet of Things」を選択します。操作するIoTドメインを選択し、「フロー・ランタイム」、「フロー・ランタイムの作成」の順に選択します。

次の値を使用します。

フィールド	値
表示名	`FR Guide - Batch Ingest`
摘要	オブジェクト・ストレージからバッチCSVファイルを読み取り、テレメトリをOCI IoTに取り込みます。
スケール	`MEDIUM`

ランタイムを作成し、アクティブになるまで待機します。

ステップ2: バッチCSVオブジェクトのアップロード

iot-data-1.csvという名前のオブジェクトをオブジェクト・ストレージ・バケットにアップロードします。
カウンタ1で始まるiot-data-<counter>.csvの形式で順次名前を使用します。

次の内容を追加します。

deviceType,externalId,time,cpuUtil,memUtil,diskUtil,firmware,temperature,humidity,pressure,mode
gateway,fr-guide-gw-01,1773768299143534,33,28,22,Oracle Linux 9.1,,,,
hvacs,fr-guide-hvac-01,1773768299144534,,,,,70.5,44.2,101.1,cool
hvacs,fr-guide-hvac-02,1773768299145534,,,,,78.3,52.0,100.7,heat

ステップ3: ノード-REDフローの構成

「IoTドメイン」リスト・ページで、フロー・ランタイムを含むドメインを選択し、「フロー・ランタイム」を選択します。フロー・ランタイム名およびオープン・フロー・ランタイム・エディタを選択します。
次の順序でノードを追加して接続します。
Inject -> Prepare Filename Function -> Object Storage -> Increment Index Function -> CSV -> Transform Function -> MQTT-OUT
1時間ごとに繰り返すようにインジェクト・ノードを構成します。
テストのために、1分間隔を使用するか、ノードを手動でトリガーできます。

filename-preparation Functionノードを1つの出力に設定し、次のコードを使用します。

let index = flow.get("iotFileIndex") || 1;

let objectName = `iot-data-${index}.csv`;

msg.fileIndex = index;
msg.objectName = objectName;
msg.payload = objectName;

return msg;

オブジェクトをダウンロードするようにオブジェクト・ストレージ・ノードを構成します:

フィールド	値
OCI構成	リソース・プリンシパル構成を作成または選択します。
工程	ダウンロード
ネームスペース	`<object-storage-namespace>`
バケット	`<bucket-name>`
オブジェクト名	空白のまま。前述の「関数」ノードは、順次オブジェクト名を指定します。
ファイル・パス	空白のまま。
出力のダウンロード	テキスト
エンコーディング	UTF-8

index-increment関数ノードを1つの出力に設定し、次のコードを使用します。

const current = msg.fileIndex || flow.get("iotFileIndex") || 1;

flow.set("iotFileIndex", current + 1);

msg.processedObjectName = msg.objectName;
return msg;

CSVノードを構成します。

フィールド	値
カラム	`deviceType,externalId,time,cpuUtil,memUtil,diskUtil,firmware,temperature,humidity,pressure,mode`
1行目のデータに列名が含まれる	有効
出力	行ごとに1つのメッセージ

変換関数ノードを1つの出力に設定し、次のコードを使用します。

const row = msg.payload || {};

const deviceType = String(row.deviceType || "").trim();
const externalId = String(row.externalId || "").trim();

if (!deviceType || !externalId) {
  node.warn("Skipping CSV row with missing deviceType or externalId");
  return null;
}

const timeValue = Number(row.time || Date.now() * 1000);

if (deviceType === "gateway") {
  msg.topic = "data";
  msg.payload = {
    time: timeValue,
    cpuUtil: Number(row.cpuUtil || 0),
    memUtil: Number(row.memUtil || 0),
    diskUtil: Number(row.diskUtil || 0),
    firmware: row.firmware || "unknown"
  };
} else if (deviceType === "hvacs") {
  msg.topic = "hvacs/" + externalId;
  msg.payload = {
    time: timeValue,
    temperature: Number(row.temperature || 0),
    humidity: Number(row.humidity || 0),
    pressure: Number(row.pressure || 0),
    mode: row.mode || "unknown"
  };
} else {
  node.warn("Unsupported CSV deviceType: " + deviceType);
  return null;
}

msg.payload = JSON.stringify(msg.payload);
return msg;

OCI IoTデバイス・ホストに公開するようにMQTT-OUTを構成します。

ノート

MQTT v5を使用する場合は、Content Type application/jsonを使用する必要があります

フィールド	値
サーバー	`<device-host>`
ポート	`8883`
TLSの使用	有効
トピック	空白のまま。「ファンクション」ノードは`msg.topic`を設定します。
QoS	`1`
ユーザ名	`fr-guide-gw-01`
パスワード	Vaultに格納されているゲートウェイデバイスのパスワード。
コンテンツ・タイプ(MQTT v5を使用している場合)	application/json

オプションで、フローの検証中に変換関数の出力をデバッグ・ノードに接続します。
エクスポートされた完全なフロー・ドキュメントをかわりにインストールするには、次のコンソール、CLIまたはAPIメソッドを使用します。

1. フロー・ランタイム・エディタを開きます。
2. ノード-REDエディタで、メイン・メニューを開き、「インポート」を選択します。
3. エクスポートしたフロー・ドキュメントを貼り付けるか、そのファイルを選択して、「インポート」を選択します。
4. インポートされたノードと接続を確認し、「デプロイ」を選択します。
完全なフロー・ドキュメントを置換するには、oci iot flow-runtime update-flowsコマンドを使用します。
```
oci iot flow-runtime update-flows \
  --iot-flow-runtime-id <flow-runtime-ocid> \
  --flows-document file://<path-to-flows-json>
```
ノート

このコマンドは、完全なフロー・ドキュメントを置き換えます。バックアップが必要な場合は、最初に現在のドキュメントを取得して保存します。

詳細は、IoTフロー・ランタイムのフローの更新を参照してください。
完全なフロー・ドキュメントを置き換えるには、UpdateIotFlowRuntimeFlows操作を実行します。
```
PUT /20250531/iotFlowRuntimes/{iotFlowRuntimeId}/flows
<complete-Node-RED-flows-document>
```
エクスポートされた完全なNode-REDフロー・ドキュメントをリクエスト本文として渡します。同期操作ではフロー・ドキュメントが返され、作業リクエストは作成されません。

ステップ4: フローのデプロイと実行

「デプロイ」を選択します
iot-data-1.csv、iot-data-2.csv、iot-data-3.csvなどの順次名を使用して、構成済バケットにCSVオブジェクトをアップロードします。
Injectノードをトリガーするか、構成された間隔を待ちます。Object Storageノードが次のオブジェクトをダウンロードしていることを確認します。
オブジェクトがダウンロードされた後にのみカウンタが増加することを確認します。次のオブジェクトが使用できない場合は、先に進めないでください。

ステップ5: バッチ・データ収集の検証

オブジェクト・ストレージ・ノードが予期したCSVオブジェクトをダウンロードし、CSVノードが行ごとに1つのメッセージを発行したことを確認します。
ゲートウェイ行がトピックdataを生成し、HVAC行がhvacs/fr-guide-hvac-01およびhvacs/fr-guide-hvac-02を生成したことを確認します。
共通コンソール検証を使用して、3つのデジタル・ツイン・インスタンスすべてについて、RAWデータまたは取込みデータ、スナップショット値、正規化値または履歴値を確認します。

トラブルシューティング

オブジェクト・ストレージ・ネームスペース、バケット名、シーケンシャル・オブジェクト名およびリソース・プリンシパル構成を確認します。
オブジェクト・ストレージ・ノードがダウンロードしたオブジェクトをバッファとして返し、index-increment関数がそれをUTF-8テキストに変換することを確認します。
次のポーリングの前にiot-data-<counter>.csvが存在することを確認します。カウンタは、ダウンロードが成功した後にのみ進みます。
CSVデリミタおよび1行目の列名がサンプルと一致することを確認します。
各行に、サポートされているdeviceType、一致する外部キーおよび数値のソース・タイムスタンプが含まれていることを確認します。
MQTTデバイス・ホスト接続、ゲートウェイ資格証明、アダプタ・マッピングおよびターゲット・デジタル・ツイン・ライフサイクルの状態を確認します。

詳細は、「IoTフロー・ランタイムのトラブルシューティング」を参照してください。

FAQ

CSVオブジェクト名でカウンタを使用するのはなぜですか。: フローは、iot-data-1.csvやiot-data-2.csvなどの予測可能な順序を処理します。カウンタは、正常にダウンロードされた後にのみ進行するため、オブジェクトは順番に処理されます。
Object Storageノードがリソース・プリンシパルを使用するのはなぜですか。: フロー・ランタイムは、フローにユーザー資格証明を格納せずに、承認済バケットにアクセスできます。動的グループおよびポリシーは、必要な読取りアクセス権を付与する必要があります。
次の順次オブジェクトが存在しない場合はどうなりますか。: ダウンロードに失敗し、カウンタは変更されません。フローを再度トリガーする前に、予期されるオブジェクトをアップロードするか、ネームスペース、バケット、接頭辞およびカウンタを修正してください。
ダウンロードされたバッファがテキストに変換されるのはなぜですか。: CSVノードにはテキスト入力が必要です。関数ノードは、CSV解析の前にオブジェクト・ストレージ・レスポンスをバッファからUTF-8テキストに変換します。

Oracle Cloud Infrastructureドキュメント