Oracle Cloud Infrastructure Documentation

Scenario: Ingesting Batch Data from Object Storage

Download sequential CSV objects from OCI Object Storage, transform each batch record in Node-RED, and ingest it into OCI IoT.

Use this scenario to replay offline or historical readings so that digital twin history remains complete. The flow is Inject -> Prepare Filename -> Object Storage Download -> Increment Index -> CSV -> Function -> MQTT-OUT.

Note

This scenario uses Object Storage because each CSV file is a discrete object that the flow downloads and processes on demand. If a flow instead needs shared files available through a mounted filesystem path, use File Storage. See Choosing File Storage or Object Storage.

Complete the common scenario setup before starting.

Prerequisites

Create an Object Storage bucket for the batch CSV objects. Record the Object Storage namespace and bucket name. Create the Flow Runtime dynamic group, and ensure that the Resource Principal configuration used by the Object Storage node has the Object Storage read policy required to download objects from the bucket.

The Flow Runtime must be able to reach Object Storage and the OCI IoT device host.

Step 1: Creating the Flow Runtime

  1. In the OCI Console, open the navigation menu. Go to Developer Services, and then select Internet of Things. Select the IoT domain you want to work with, select Flow Runtimes, and then select Create Flow Runtime.
  2. Use the following values:
    FieldValue
    Display nameFR Guide - Batch Ingest
    DescriptionReads batch CSV files from Object Storage and ingests telemetry into OCI IoT.
    ScaleMEDIUM
  3. Create the runtime and wait for it to become active.

Step 2: Uploading the Batch CSV Object

  1. Upload an object named iot-data-1.csv to the Object Storage bucket.

    Use sequential names in the format iot-data-<counter>.csv, beginning with counter 1.

  2. Add the following content:
    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

Step 3: Configuring the Node-RED Flow

  1. On the IoT domains list page, select the domain with the Flow Runtime, and then select Flow runtimes. Select the Flow Runtime name and Open Flow runtime editor.
  2. Add and connect these nodes in order:

    Inject -> Prepare Filename Function -> Object Storage -> Increment Index Function -> CSV -> Transform Function -> MQTT-OUT

  3. Configure the Inject node to repeat every hour.

    For testing, you can use a one-minute interval or trigger the node manually.

  4. Set the filename-preparation Function node to one output and use this code:
    let index = flow.get("iotFileIndex") || 1;

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

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

return msg;
  5. Configure the Object Storage node to download the object:
    FieldValue
    OCI ConfigCreate or select a Resource Principal configuration.
    OperationDownload
    Namespace<object-storage-namespace>
    Bucket<bucket-name>
    Object nameLeave blank. The preceding Function node supplies the sequential object name.
    File pathLeave blank.
    Download outputBuffer
  6. Set the index-increment Function node to one output and use this code:
    if (Buffer.isBuffer(msg.payload)) {
  msg.payload = msg.payload.toString("utf8");
}

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

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

msg.processedObjectName = msg.objectName;
return msg;
  7. Configure the CSV node:
    FieldValue
    ColumnsdeviceType,externalId,time,cpuUtil,memUtil,diskUtil,firmware,temperature,humidity,pressure,mode
    First row contains column namesEnabled
    OutputOne message per row
  8. Set the transformation Function node to one output and use this code:
    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;
  9. Configure MQTT-OUT to publish to the OCI IoT device host:
    Note

    If you use MQTT v5, then you must use Content Type application/json
    FieldValue
    Server<device-host>
    Port8883
    Use TLSEnabled
    TopicLeave blank. The Function node sets msg.topic.
    QoS1
    Usernamefr-guide-gw-01
    PasswordGateway device password stored in Vault.
    Content Type (if using MQTT v5) application/json
  10. Optionally, connect the transformation Function output to a Debug node while validating the flow.
  11. To install an exported complete flows document instead, use the following Console, CLI, or API method.
    1. Open the Flow Runtime editor.
    2. In the Node-RED editor, open the main menu and select Import.
    3. Paste the exported flows document or select its file, and then select Import.
    4. Review the imported nodes and connections, and then select Deploy.

  • Use the oci iot flow-runtime update-flows command to replace the complete flows document.

    oci iot flow-runtime update-flows \
  --iot-flow-runtime-id <flow-runtime-ocid> \
  --flows-document file://<path-to-flows-json>
    Note

    The command replaces the complete flows document. Retrieve and save the current document first if you need a backup.

    For more information, see Updating Flows for an IoT Flow Runtime.

  • Run the UpdateIotFlowRuntimeFlows operation to replace the complete flows document.

    PUT /20250531/iotFlowRuntimes/{iotFlowRuntimeId}/flows
<complete-Node-RED-flows-document>

    Pass the exported complete Node-RED flows document as the request body. The synchronous operation returns the flows document and doesn't create a work request.

Step 4: Deploying and Running the Flow

  1. Select Deploy.
  2. Upload CSV objects to the configured bucket using sequential names such as iot-data-1.csv, iot-data-2.csv, and iot-data-3.csv.
  3. Trigger the Inject node or wait for its configured interval. Confirm that the Object Storage node downloads the next object.
  4. Confirm that the counter increments only after an object is downloaded. It must not advance when the next object isn't available.

Step 5: Validating the Batch Data Ingestion

  1. Confirm that the Object Storage node downloaded the expected CSV object and the CSV node emitted one message for each row.
  2. Confirm that the gateway row produced topic data and the HVAC rows produced hvacs/fr-guide-hvac-01 and hvacs/fr-guide-hvac-02.
  3. Use common Console validation to confirm raw or ingested data, snapshot values, and normalized or historical values for all three digital twin instances.

Troubleshooting

  • Confirm the Object Storage namespace, bucket name, sequential object name, and Resource Principal configuration.
  • Confirm that the Object Storage node returns the downloaded object as a Buffer and that the index-increment Function converts it to UTF-8 text.
  • Confirm that iot-data-<counter>.csv exists before the next poll. The counter advances only after a successful download.
  • Confirm that the CSV delimiter and first-row column names match the sample.
  • Confirm that each row includes a supported deviceType, a matching external key, and a numeric source timestamp.
  • Confirm the MQTT device-host connection, gateway credentials, adapter mappings, and target digital twin lifecycle states.

For more information, see Troubleshooting IoT Flow Runtimes.

FAQs

Why do the CSV object names use a counter?
The flow processes a predictable sequence such as iot-data-1.csv and iot-data-2.csv. The counter advances only after a successful download so objects are processed in order.
Why does the Object Storage node use a Resource Principal?
The Flow Runtime can access the approved bucket without storing user credentials in the flow. Its dynamic group and policy must grant the required read access.
What happens when the next sequential object doesn't exist?
The download fails and the counter remains unchanged. Upload the expected object or correct the namespace, bucket, prefix, and counter before triggering the flow again.
Why is the downloaded Buffer converted to text?
The CSV node expects text input. The Function node converts the Object Storage response from a Buffer to UTF-8 text before CSV parsing.