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 download sequential CSV objects from Object Storage and replay offline or historical readings into OCI IoT so that digital twin history remains complete. The Node-RED flow is Inject -> Prepare Filename -> Object Storage Download -> Increment Index -> CSV -> Transform -> 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 IoT 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

Choose an Import or Manual Configuration Path

You can complete this batch ingestion scenario in two ways, both approaches show how to download CSV objects from Object Storage and ingest the batch data into OCI IoT:
  1. Option 1: Complete Step 3 to import this scenario's example JSON flows file into the managed Node-RED editor, modify the imported flow to use your resources, and review its configuration.
  2. Option 2: Go to Step 4 to build and configure the flow directly in the managed Node-RED editor.
Note

If you already built this scenario manually, you do not need to import the JSON. Do not deploy both the imported and manually built copies; they can ingest duplicate readings.

Step 3: Importing this Scenario's JSON Flows file

The managed Node-RED editor opened from the OCI Console includes the OCI nodes required in this example.

The import creates a flow tab named Ingest Batch Data from Object Storage and its configuration nodes. It doesn't create the bucket, access policies, or digital twin resources required by the preceding steps and common setup. MQTT credentials aren't included in the exported JSON.

  1. Download ingesting-batch-data-from-object-storage-flow.json to your computer.

    If the browser displays the JSON, use its save option to save the file with the .json extension.

  2. In the OCI Console, select your IoT domain, select Flow runtimes, open the runtime created in Step 1, and select Open Flow runtime editor.
  3. In the editor's main menu, select Import, choose the downloaded JSON file, select the option to import into a new flow, and select Import.
  4. Replace the four placeholder values in the imported node configurations using the following table.

    Double-click a node to open its editor. For a configuration shared by nodes, select the edit button next to its configuration field. Replace the entire placeholder, including the angle brackets, with your value. For example, replace <bucket-name> with your actual bucket name, such as iot-batch-data. Enter values without quotation marks. Save each configuration and select Done to close the node editor.

    PlaceholderWhere to replace itValue to enter
    <region>Open the Object Storage node, edit its OCI Config, and set the region.The OCI region identifier for the bucket, such as us-ashburn-1. Use the region where you created the bucket, rather than its display name. Keep Resource Principal authentication selected.
    <object-storage-namespace>Object Storage node: Namespace.Copy the Object Storage namespace from the bucket's details in the OCI Console. This is the namespace recorded in the prerequisites, not the tenancy display name or an OCID.
    <bucket-name>Object Storage node: Bucket.Copy the exact name of the bucket containing iot-data-1.csv from Step 2. Preserve its capitalization; don't enter the bucket OCID or a URL.
    <device-host>Open the MQTT-OUT node, edit its server configuration, and replace Server.Copy Device host from the IoT domain details. See Getting an IoT Domain's Details. Enter the hostname only, without a URL scheme, port, or path. The port is configured separately as 8883.
  5. In the MQTT server configuration, enter the gateway credentials and review the connection settings.

    On the Security tab, enter the gateway external key as Username: fr-guide-gw-01 if you used the common setup values. Enter the actual gateway device password stored in Vault as Password, not the secret OCID or your OCI Console password. These values come from creating the gateway digital twin instance.

    For Client ID, choose a value unique among clients connected to this device host, such as fr-guide-batch-01; use a different value if you deploy another copy. Review the TLS and other settings in the MQTT-OUT configuration step, and retain MQTT 5.0 with Content Type set to application/json.

  6. Use the following links to review the imported nodes against the scenario instructions before deploying.

    The nodes are already connected. Review their existing settings and code; you don't need to add duplicate nodes.

    Imported nodeScenario instructions
    InjectReview the hourly interval and manual trigger.
    Build file nameReview how the next sequential object name is constructed.
    Object StorageReview Resource Principal access and Text/UTF-8 download settings.
    Index IncrementReview the file counter update after download.
    CSVReview the columns and output of one message per row.
    csv to IoTReview the gateway and HVAC topic and payload mappings.
    MQTT-OUTReview the device-host connection and authentication.
    Catch, OS Node error catch, and MQTT InputReview error reporting and the messages sent to MQTT-OUT.

Step 4: Reviewing and Configuring the Node-RED Flow

If you imported the example, review and adjust the existing nodes using these steps. If you are building the flow manually, add and connect the nodes before configuring them.

  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. If you are building the flow manually, add and connect these nodes in order:

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

    For instructions on replacing the example placeholders with your own resource values, see Step 3: Importing this Scenario's JSON Flows file.

  3. Configure the Inject node to repeat every hour.

    The imported flow already uses an hourly interval and doesn't inject once at startup. For the first test, set the repeat option to none and trigger the node manually. Restore the hourly interval after validation.

  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 outputText
    Encoding UTF-8
  6. Set the index-increment Function node to one output and use this code:
    const 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. Review the Catch and Debug nodes while validating the flow.

    The imported Catch node is scoped to the Object Storage node and sends errors to the OS Node error catch Debug node. The MQTT Input Debug node shows the transformation output sent to MQTT-OUT; it doesn't confirm that OCI IoT received the message. If building manually, add these connections to inspect errors and transformed messages in the Debug sidebar.

Working with an Exported Complete Flows JSON file

  • To import the downloadable scenario, use Step 3: Importing this Scenario's JSON Flows file. For another exported flows document, use these editor steps. The CLI and API methods replace the runtime's complete flows document.

    1. Open the Flow Runtime editor. If you need help finding the editor, see Getting an IoT Flow Runtime's Details.
    2. In the Node-RED editor, open the main menu and select Import. For more details, see the Node-RED import documentation.
    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.

    Note

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

Step 5: Deploying and Running the Flow

  1. Confirm that the resource placeholders have been replaced, gateway credentials have been entered, and the nodes show no configuration errors. 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. For the first test, trigger the Inject node manually. Confirm that the Object Storage node downloads the next object.

    A new flow begins with iot-data-1.csv. Each successful download advances iotFileIndex in flow context. Subsequent triggers use the next filename, including when you repeat a test.

  4. Confirm that the counter increments only after an object is downloaded. It must not advance when the next object isn't available.

Step 6: 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.
  4. After validation, restore the Inject node's hourly interval and select Deploy.

    Keep the next sequential CSV object available before each scheduled run. You can disable the two Debug nodes using their enable/disable buttons after validation, while keeping the Catch node and its connections for later troubleshooting.

Troubleshooting

  • Confirm the Object Storage namespace, bucket name, sequential object name, and Resource Principal configuration.
  • Confirm that the Object Storage node uses Download output = Text and Encoding = UTF-8. The CSV node receives text directly.
  • Review Object Storage errors in OS Node error catch and transformed messages in MQTT Input in the Debug sidebar. Re-enable the Debug nodes if you disabled them after validation.
  • 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 does the Object Storage node download the object as text?
The CSV node expects text input. Setting the download output to Text with UTF-8 encoding supplies that input directly; the index-increment Function only advances the counter and records the processed object name.
How do I restart a test from the first CSV object?
Disable the Inject node's repeat interval, then add a temporary Function node on the same flow tab with flow.set("iotFileIndex", 1); return msg;. Connect a separate, manually triggered Inject node only to that Function node. Deploy and trigger it once to reset the counter, then remove both temporary nodes and deploy again. Trigger the scenario's Inject node to process iot-data-1.csv again. This replays readings already ingested; use it only when you intend to repeat the test.