Load entity data with a pipeline

The previous lesson defined the structure of iot_machine. Now create a source dataset and load two rows through a Batch pipeline's Entity output. Keep the entity and dataset in the same collection so both are available in the component library.

Prerequisites

• The iot_machine entity from the previous lesson
• id configured as its Identity Key
• Permission to create datasets and pipelines in the practice collection

Create the machines source dataset

1. Open the practice collection from Collections.
2. Select Add item → Dataset → Table.
3. Enter the name machines under Basic information, then select Next.
4. Create this schema.

5. Select Create.
6. Download machines.csv, or save the following content with that filename.

id,sensor_count,reading_count,latest_recorded_at,anomaly_count,latest_health_score
machine_001,1,120,2026-07-01T09:00:00,2,0.92
machine_002,2,80,2026-07-01T09:05:00,0,0.99

7. Open machines and add the CSV under Data → Upload.
8. Confirm that Data shows two rows.

The source stores latest_recorded_at as Text while the entity uses Timestamp. The pipeline Code converts that type.

Create a Batch pipeline

1. Open Data → Pipelines.
2. Select Create.
3. In Select pipeline collection, choose the collection containing machines and iot_machine.
4. Keep the pipeline type set to Batch.

Create Python Code

1. Drag Quick add → Code → Python from the component library to the canvas.
2. Enter the name map_machines_to_iot_machine.
3. Use this code.

import polars as pl

def run(machines, *, options=None, contexts=None):
    output = machines.with_columns(
        pl.col("id").cast(pl.Utf8),
        pl.col("sensor_count").cast(pl.Int64),
        pl.col("reading_count").cast(pl.Int64),
        pl.col("latest_recorded_at").str.to_datetime(strict=False),
        pl.col("anomaly_count").cast(pl.Int64),
        pl.col("latest_health_score").cast(pl.Float64),
    )
    return {"iot_machine": output}

The runtime passes pipeline input keys as keyword arguments. Input key machines must match function argument machines, and returned key iot_machine must match the output key.

Connect input, Code, and Entity

1. Add machines from the collection section of the component library.
2. Connect its right handle to the Code node's left handle.
3. Add the iot_machine entity displayed as CNC machine.
4. Connect the Code node's right handle to iot_machine.
5. In the Code node's Options, confirm input key machines and output key iot_machine.
6. Confirm write mode Upsert on the Entity output.

The pipeline has one source, one transformation, and one Entity output.

flowchart LR
    accTitle: Pipeline from source dataset to entity
    accDescr: Python Code transforms the machines source dataset to the entity schema and upserts the result into iot_machine.
    dataset[(machines<br/>source dataset)] -->|input key machines| code[map_machines_to_iot_machine<br/>Python Code]
    code -->|output key iot_machine| entity[(iot_machine<br/>Entity Upsert)]

Save and run

1. Select Save and name the pipeline load_iot_machines.
2. Select Run now.
3. Wait for top-level status Ready and latest run Success.
4. Confirm that step map_machines_to_iot_machine also succeeded.

Upsert updates an existing row when the same id arrives again. Running the pipeline again must not create duplicate machine_001 or machine_002 rows.

Verify the Data tab

1. Return to Ontology → Modeling.
2. Select the CNC machine node in the practice collection.
3. Open Data and refresh if needed.
4. Confirm these two rows.

Graph synchronization runs in the background, so the nodes may appear in Graph Explorer shortly afterward.

Self-check

• Two rows are uploaded to machines.
• Input key and function argument are both machines.
• Return key and Entity output key are both iot_machine.
• Write mode is Upsert.
• Status is Ready and the latest run is Success.
• Two rows appear under iot_machine Data.

Next lesson

Next, define a relation from iot_sensor to iot_machine and load its structural columns through a Relation output.