Connecting Industrial Assets to a Knowledge Graph Using MQTT, Node-RED, and Neo4j

In the previous article, we built a knowledge graph for manufacturing operations.

This article is the how. We're going to build a small but complete pipeline that takes an MQTT message published by a shop-floor system, transforms it into a Cypher write, and updates the knowledge graph — automatically, in real time, every time the message arrives.

The pieces:

• MQTT as the transport — operational systems publish events to topics, the knowledge graph subscribes to them.
• Node-RED as the mapping layer — receiving the events, validating and shaping them, generating the Cypher.
• Neo4j as the destination — the same knowledge graph from the previous article, now hydrated dynamically.

A note on scope: this article is the integration pattern, not the architecture. The broader question of how to design a proper Unified Namespace — topic taxonomy, broker topology, data contracts, governance, Sparkplug B — is what the next series tackles. Here we keep it tight: one broker, two mapping patterns, enough to prove the loop closes.

What we're going to build

The pipeline subscribes to topics the plant's asset and control layer publishes to. The topic structure is the plant's ISA-95 path, which matters more than it looks — it means a single subscription with wildcards serves every line at every site:

The five wildcard segments are enterprise / site / area / work-cell, so gv1.0/GlobalIndustries/Munich/PackagingArea/Line1/EquipmentDefinition/EQ-MUC-FL501 is a fully-qualified address for one asset on one line.

These two were chosen because they exercise the fundamental mapping patterns you'll meet in any UNS-to-knowledge-graph integration: upsert a classified instance (commissioning), and append an immutable, history-bearing event (production events). Handle these two and you can handle most of what a real plant throws at a graph.

The pipeline in one picture

Each flow is the same shape end to end:

MQTT broker              Node-RED                          Neo4j
-----------              --------------------------        ------
gv1.0/…/EquipmentDefinition  MQTT in (JSON)                HTTP transactional
                         --> Function: build Cypher Tx     endpoint
gv1.0/…/Production/Events    Function: check response  --> /db/neo4j/tx/commit
                         --> Debug

One flow per message family — each with its own MQTT-in node, a function node that builds a parameterised Cypher transaction, an HTTP request node that POSTs to Neo4j's transactional endpoint, and a second function node that checks the response. We use the HTTP endpoint rather than a contrib package so the flows have zero extra Node-RED dependencies beyond what ships standard.

Pattern 1: Equipment commissioning

When a piece of equipment is commissioned (or its master data changes), the asset model publishes its full definition. The Munich filler arrives as:

Payload:

{
  "schemaVersion": "1.0.0",
  "equipment": {
    "id": "EQ-MUC-FL501",
    "assetTag": "FL-501",
    "name": "Rotary Piston Filler FL-501",
    "lifecycleStatus": "commissioned",
    "serialNumber": "KRN-MF12-18-0912",
    "installationDate": "2018-11-05"
  },
  "equipmentClass": { "id": "EQCLASS-FILLER", "name": "Filler" },
  "equipmentDefinition": {
    "id": "EQDEF-KRN-MODULFILLHRS12",
    "manufacturer": "Krones",
    "model": "Modulfill HRS 12",
    "description": "12-head rotary piston hot-fill filler, 200 BPM, fills at >82C for FDA acid-food pasteurization"
  },
  "parent": { "workCenterId": "WC-MUC-300-L1" },
  "source": {
    "origin": "BatchPlantSimulator_OPCUA",
    "uri": "opc.tcp://localhost:26543/BatchPlantServer/Area500_Filling/Filler_FL501",
    "gateway": "Ignition-Gateway-001",
    "originSystemId": 1015
  },
  "quality": "Good",
  "timestamp": 1779277854779
}

The single most useful field here is equipmentClass.name. It says "Filler" — and Filler is a real class in the production ontology, sitting under the core Equipment. The payload is already speaking the ontology's vocabulary; the mapping's job is to connect the two.

MERGE (eq:Instance:Equipment {id: $eqId})
ON CREATE SET eq.created_at = datetime()
SET eq.assetTag            = $eqAssetTag,
    eq.name                = $eqName,
    eq.equipmentType       = $eqClassName,
    eq.lifecycleStatus     = $eqLifecycleStatus,
    eq.serialNumber        = $eqSerialNumber,
    eq.installationDate    = date($eqInstallationDate),
    eq.manufacturer        = $eqDefManufacturer,
    eq.model               = $eqDefModel,
    eq.description         = $eqDefDescription,
    eq.equipmentClassId    = $eqClassId,
    eq.equipmentDefinitionId = $eqDefId,
    eq.sourceOrigin        = $sourceOrigin,
    eq.sourceUri           = $sourceUri,
    eq.sourceGateway       = $sourceGateway,
    eq.sourceSystemId      = $sourceSystemId,
    eq.schemaVersion       = $schemaVersion,
    eq.lastSeenAt          = datetime({epochMillis: $timestamp}),
    eq.lastSeenQuality     = $quality
WITH eq

// Type the asset against its concrete ontology class, e.g. 'Filler'.
OPTIONAL MATCH (cls:Class {name: $eqClassName})-[:SUBCLASS_OF*]->(:Class {name: 'Equipment'})
FOREACH (c IN CASE WHEN cls IS NULL THEN [] ELSE [cls] END |
    MERGE (eq)-[:IS_INSTANCE_OF]->(c))
WITH eq, cls

// Place the asset at its work centre.
OPTIONAL MATCH (wc:Instance:WorkCenter {id: $parentWorkCenterId})
FOREACH (w IN CASE WHEN wc IS NULL THEN [] ELSE [wc] END |
    MERGE (eq)-[:IS_LOCATED_AT]->(w))

RETURN eq.id AS equipmentId,
       cls.name AS typedAs,
       wc.id AS locatedAt,
       CASE WHEN cls IS NULL
            THEN 'no ontology class named "' + $eqClassName + '" under Equipment — asset is untyped'
            ELSE null END AS classWarning,
       CASE WHEN wc IS NULL
            THEN 'no work centre instance "' + $parentWorkCenterId + '" — asset is unplaced'
            ELSE null END AS locationWarning

Four design points worth pulling out.

One node, not three. An earlier version of this pipeline created three instance nodes — an EquipmentClass, an EquipmentDefinition and the Equipment itself — mirroring an ontology that had all three tiers. The current ontology doesn't: Equipment is a core class specialised directly into concrete types like Filler and CookKettle. So the mapping types the asset against its leaf class and keeps the nameplate as properties. The class and definition ids are retained as equipmentClassId and equipmentDefinitionId so the original grouping is recoverable if a definition tier is ever re-modelled — data preserved, structure not invented.

IS_LOCATED_AT, not CONTAINS. It's tempting to reach for CONTAINS, since that's how the plant topology nests. But CONTAINS is declared Location → Location — it's how a site contains an area. Equipment is a PhysicalEntity, and the relationship for putting a physical thing somewhere is IS_LOCATED_AT, whose declared range is Location. The ontology's domain and range aren't decoration; they tell you which edge to write.

The class lookup is constrained to descendants of Equipment. Without -[:SUBCLASS_OF*]->(:Class {name: 'Equipment'}), a payload with equipmentClass.name of "Site" would happily type a filler as a location. The traversal makes that impossible.

Both lookups are OPTIONAL, and both report. This is the important one, and it's covered in its own section below.

Every node is MERGEd on a stable, upstream-supplied id — EQ-MUC-FL501. Re-publishing the same commissioning message (which MQTT QoS 1 explicitly allows) finds the existing node instead of creating a duplicate. The id comes from the source, not from Neo4j's internal ids, precisely so idempotency survives redelivery.

The failure mode that hides in plain sight

The earlier version of this Cypher looked reasonable:

MERGE (eq:Instance:Equipment {id: $eqId})
SET   eq.assetTag = $eqAssetTag, ...
WITH  eq
MATCH (cEq:Class {name: 'Equipment'})
MATCH (cEd:Class {name: 'Equipment Definition'})
MERGE (eq)-[:IS_INSTANCE_OF]->(cEq)
WITH  eq
MATCH (wc:Instance:WorkCenter {id: $parentWorkCenterId})
MERGE (eq)-[:IS_PART_OF]->(wc)
RETURN eq.id AS equipmentId

When the ontology was rebuilt, Equipment Definition ceased to exist. That MATCH stopped matching — and in Cypher, a MATCH that finds nothing eliminates the row, so every clause after it silently does nothing.

The result was not an error. It was worse. The MERGE above the failing MATCH still ran, so every message created an Equipment node. Nothing below it ran, so the node was never typed, never placed, and the statement returned no rows. Untyped, orphaned equipment accumulated in the graph on every publish, and the only outward sign was a debug pane showing an empty result — which looks a lot like nothing happening.

An earlier draft of this article described that hard MATCH as "a deliberate fail-safe: equipment can only attach to topology that's actually been provisioned." That reasoning is wrong, and worth correcting explicitly, because it's seductive. A hard MATCH doesn't fail safe. It fails silently, and it takes the rest of the statement with it.

The fix is OPTIONAL MATCH plus a FOREACH guard to do the conditional write, and — critically — returning a diagnostic:

RETURN eq.id AS equipmentId,
       cls.name AS typedAs,
       wc.id AS locatedAt,
       CASE WHEN cls IS NULL THEN 'asset is untyped'  ELSE null END AS classWarning,
       CASE WHEN wc  IS NULL THEN 'asset is unplaced' ELSE null END AS locationWarning

Now a missing work centre still commissions the asset, and says so. The response-checking node turns those warnings into a yellow status and a node.warn rather than a green tick:

const [equipmentId, typedAs, locatedAt, classWarning, locationWarning] = row;
const warnings = [classWarning, locationWarning].filter(Boolean);
if (warnings.length > 0) {
    node.warn(equipmentId + ': ' + warnings.join('; '));
    node.status({ fill: 'yellow', shape: 'ring', text: equipmentId + ' (' + warnings.length + ' warning)' });
} else {
    node.status({ fill: 'green', shape: 'dot', text: equipmentId + ' → ' + typedAs + ' @ ' + locatedAt });
}

The principle generalises well beyond this flow: in an ingestion path, the difference between "nothing matched" and "everything worked" must never be invisible.

Pattern 2: Production events

The second pattern appends immutable event nodes. Order, work-order, batch and phase events all arrive on gv1.0/…/Production/Events/# and share a shape:

{
  "eventType": "WorkOrderActivated",
  "eventId": "EVT-WO-ACT-26154-0007",
  "timestamp": "2026-06-03T06:15:00+02:00",
  "correlationId": "workorder-WO-MUC-26154-0007",
  "source": { "origin": "MES", "gateway": "Ignition-Gateway-MUC-Line1", "originSystemId": 2001 },
  "data": {
    "workOrderId": "WO-MUC-26154-0007",
    "processOrderNumber": "1000234567",
    "state": "ACTIVE",
    "isa95Context": { "enterprise": "GlobalIndustries", "site": "Munich",
                      "area": "ProductionArea", "processCell": "Line1" }
  }
}

The mapping upserts the entity the event concerns, upserts the event itself, and links them:

MERGE (wo:Instance:WorkOrder {id: $entityId})
  ON CREATE SET wo.created_at = datetime()
  SET wo.materialNumber = $materialNumber, wo.producedBatch = $producedBatch
MATCH (woCls:Class {name: 'WorkOrder'})
MERGE (wo)-[:IS_INSTANCE_OF]->(woCls)
MERGE (evt:Instance:ProductionEvent {id: $eventInstanceId})
  ON CREATE SET evt.created_at = datetime()
  SET evt.eventType = $eventType, evt.eventTime = datetime({epochMillis: $eventTime})
MERGE (evt)-[:CONCERNS]->(wo)

Note the class names have no spaces — WorkOrder, ProcessOrder, ControlRecipe, ProductionEvent. An earlier ontology used Work Order and Process Order; the rebuild uses PascalCase throughout. Those five names were exactly the silent-failure case described above, and correcting them is what took this flow from writing nothing to populating the whole order cascade.

The event id is derived from the source's eventId, so redelivery finds the existing event node instead of appending a duplicate.

Asserting state from events

An event usually means something changed. The ontology has a specific vocabulary for that, and it's worth understanding because the naive approach quietly loses information.

Four relationships work together:

(workOrder)-[:HAS_STATE]->(ActiveState)          current condition — exactly one edge
(event)-[:CHANGES_STATE]->(workOrder)            what changed
(event)-[:TRANSITIONS_TO]->(ActiveState)         what it became
(event)-[:TRANSITIONS_FROM]->(DispatchedState)   what it left

The entity carries one HAS_STATE edge, replaced on each change. History is not accumulated on the entity — it lives in the event chain, because every event carries a timestamp and the state it produced:

EVT-1  t=1000   —                → ReceivedState
EVT-2  t=2000   ReceivedState    → DispatchedState
EVT-3  t=3000   DispatchedState  → ActiveState

The Cypher has one trap, and it's the same trap as before:

MERGE (evt)-[:CHANGES_STATE]->(wo)
MATCH (toS:Class {name: $toState})
MERGE (evt)-[:TRANSITIONS_TO]->(toS)
WITH wo, evt, toS
OPTIONAL MATCH (wo)-[:HAS_STATE]->(prev:Class)
FOREACH (p IN CASE WHEN prev IS NULL THEN [] ELSE [prev] END |
  MERGE (evt)-[:TRANSITIONS_FROM]->(p))
WITH wo, toS
OPTIONAL MATCH (wo)-[old:HAS_STATE]->()
DELETE old
WITH wo, toS
MERGE (wo)-[:HAS_STATE]->(toS)

That OPTIONAL MATCH (wo)-[old:HAS_STATE]->() before the DELETE is not defensive style — it is load-bearing. Written as a plain MATCH, the very first state change for an entity finds no existing edge, the row is eliminated, and the MERGE that sets the new state never runs. The state silently never gets written, and every subsequent event fails the same way.

Why keep state on the entity at all, rather than a reified state-assertion node with validity periods? Because the event node already carries everything such an assertion would hold — timestamp, source system, gateway, originating system id, data quality. A separate assertion node would duplicate all of it and add only VALID_UNTIL, which is the next event's timestamp. Where a regulated audit trail demands more — authority, reason, signature — the reified assertion is the right escalation. It just isn't the default.

Durations and time-in-state are deliberately absent from the graph. They belong in the historian, for the same reason metric values do: the graph holds structure and current condition; the time-series store holds the series.

The Node-RED function: building the transaction

Each flow's first function node validates the payload and assembles the HTTP request body:

const p = msg.payload;

const required = ['equipment', 'equipmentClass', 'equipmentDefinition', 'parent'];
for (const k of required) {
    if (!p || !p[k]) { node.error('Payload missing required section: ' + k, msg); return null; }
}
if (!p.equipment.id || !p.equipmentClass.name || !p.parent.workCenterId) {
    node.error('Payload missing required fields', msg); return null;
}

msg.payload = { statements: [{ statement: cypher, parameters: { /* … */ } }] };
msg.headers = { 'Content-Type': 'application/json' };
return msg;

Two notes. The function picks fields off the payload explicitly — anything it doesn't reference (the enumeration array some publishers include, for instance) is silently dropped, which is exactly what you want; the ontology is the authoritative enumeration, not the payload. And required-field validation lives here, in the mapping layer, dropping malformed messages with a loud node.error rather than letting them reach Neo4j.

Note that equipmentClass.name is now a required field. It drives the ontology typing, so a message without it produces an untyped asset — better to reject it at the boundary.

The second function node exists because of a Neo4j quirk worth knowing:

// Neo4j's HTTP Tx endpoint returns HTTP 200 even on Cypher errors —
// the errors array in the body is the real signal.
const body = msg.payload;
if (body && body.errors && body.errors.length > 0) {
    node.error('Neo4j returned errors: ' + JSON.stringify(body.errors), msg);
    return null;
}
const row = body && body.results && body.results[0] && body.results[0].data
         && body.results[0].data[0] && body.results[0].data[0].row;
if (!row) {
    node.error('Statement returned no rows — nothing was written', msg);
    return null;
}

A 200 OK from the transactional endpoint does not mean the write succeeded — a Cypher error comes back inside the errors array of an otherwise-200 response. And an empty results array, as the silent-failure section showed, means the statement matched nothing. Both need catching.

Getting it running

1. Import the flows into Node-RED (Menu → Import).
2. Open the MQTT broker config node and point it at your broker.
3. Open the HTTP request node, set the URL and Basic Auth credentials for your Neo4j.
4. Deploy.
5. Publish a test message and watch the debug pane.

A smoke test for the commissioning flow:

mosquitto_pub -h your-broker -p 1883 -u <user> -P <pass> \
  -t 'gv1.0/GlobalIndustries/Munich/PackagingArea/Line1/EquipmentDefinition/EQ-MUC-FL501' \
  -f EquipmentDefinition.json

If the ontology, the constraints, and the plant topology are loaded, you'll see a new Equipment node typed as Filler and located at Line 1, and the function node's status flips to green with EQ-MUC-FL501 → Filler @ WC-MUC-300-L1. The graph just hydrated itself from an event.

Confirm it end to end:

MATCH (eq:Instance:Equipment {id:'EQ-MUC-FL501'})-[:IS_LOCATED_AT]->(wc)
MATCH (eq)-[:IS_INSTANCE_OF]->(c)-[:SUBCLASS_OF*]->(root)
WHERE NOT (root)-[:SUBCLASS_OF]->()
MATCH p=(:Instance:Enterprise)-[:CONTAINS*]->(wc)
RETURN [n IN nodes(p) | n.name] AS placement, c.name AS typedAs, root.name AS rootConcept;

placement                                                     typedAs   rootConcept
["Global Industries","Munich Plant","Packaging Area","Line 1"] "Filler"  "PhysicalEntity"

One MQTT message, and the asset is connected to both the plant's physical topology and the semantic model.

The lessons that cost real time

The flows above look tidy. Getting them to that state surfaced a series of failures worth documenting, because every one of them is a thing the next person will hit.

A failed MATCH silently voids the rest of the statement. Covered above, but it earns first place: it cost more time than everything else combined, and it recurred three separate times in this pipeline — the class lookups, the work-centre lookup, and the state-edge replacement. If a clause might not match, OPTIONAL MATCH it and report the miss.

Neo4j returns HTTP 200 on Cypher errors. The errors array in the body is the real signal. Without the checking node, failed writes look like successes.

The repo and the running container drift apart. The flows live in flows.json, but Node-RED runs from its own working directory. If that directory is a named Docker volume, the file in the repo is only a seed — Docker copies an image's contents into a newly created volume and never touches an existing one. Editing the repo and restarting appears to work and changes nothing. The symptom is maddening: verified fixes that don't take effect. The fix is to bind-mount the data directory so the repo file is the runtime file, and there is only ever one copy.

Uniqueness constraints, before any ingest. During iterative testing — republishing equipment as an Equipment Class was renamed — the graph accumulated duplicates. MERGE is only idempotent against identical input; it can't protect you from property drift across iterations. Run the constraints first and Neo4j refuses the second node outright.

An empty required field silently drops the write. A piece of equipment wouldn't ingest; the culprit was an empty parent.workCenterId in the source asset model. Empty string is falsy, so validation correctly rejected it. The loud-failure validation was doing its job; the fix belonged at the source.

Non-ASCII characters arrive mangled. A manufacturer field of "Bühler" showed up with a Unicode replacement character. The source wasn't emitting UTF-8. Fix the publisher's encoding.

None of these are reasons not to use this pattern. They're the cost of admission, paid once, documented here so you pay it faster.

What this pipeline gets right — and what it doesn't

What it gets right:

The mapping is ontology-aligned. The payload already names its type — "Filler" — and the mapping resolves that against the ontology rather than hard-coding a translation table. Add a new equipment type to the production ontology and the pipeline handles it with no code change.

The pipeline is idempotent. Instances MERGE on stable upstream ids; events are keyed on the source's event id; uniqueness constraints backstop both. Redelivered messages, replayed events and broker reconnects don't corrupt the graph.

Failures are visible. Every lookup that might miss reports that it missed. This is the property the first version lacked, and its absence was invisible precisely because the failure produced no error.

What it doesn't get right — yet:

The state-assertion pattern is declared but not yet wired into every flow. The ontology defines HAS_STATE, CHANGES_STATE, TRANSITIONS_TO and TRANSITIONS_FROM, and the Cypher above is proven, but the production-event flow still records state as a string property on the entity. Migrating it is mechanical and unglamorous, and it is the difference between a graph that knows an order is active and one that can only tell you a string it was sent.

It assumes a single broker and a single Neo4j, both reachable directly. A real plant has segmented networks, redundant brokers, and security boundaries. That's architecture, not mapping — the next series.

There's no schema validation beyond the function node's required-field check. Production deployments want a JSON-schema gate (or Sparkplug B's implicit protobuf schema) dropping non-conforming messages to a dead-letter topic.

The Cypher lives in the function nodes. Fine for a handful; at thirty you'd externalise the templates.

And the non-equipment master-data feeds aren't built. Materials, personnel and suppliers each need their own flow following this same pattern, fed by ERP, LIMS and MES.

What's next

What we've built is one strand of the dynamic layer — one broker, a couple of message families, hydrating one knowledge graph with the equipment and events that change fastest. It works, and it answers the question the previous article left hanging: how you keep the graph current without a human typing Cypher.

But this is the smallest version of the answer. A real plant has dozens of source systems — SCADA, MES, ERP, LIMS, CMMS, historian, lab — each with its own data model, cadence and reliability profile. Coordinating all of them onto a single namespace that the knowledge graph and every other consumer can subscribe to is the architecture problem the next series is built around.

The next article starts that series. We'll step back from the wire-up details and look at what a Unified Namespace actually is, why it's the backbone of a modern manufacturing data architecture, and how it sits in relation to the ontology and knowledge graph you've just built. The flows you've imported are clients on that namespace; the next series is about everything that has to be true upstream of them for those clients to be useful at scale.