Datensätze ausrichten und korrelieren

Ordnen Sie verwandte Werte zu, die als separate Nachrichten ankommen. Diese Beispiele verwenden eine MQTT-Quelle mit mehreren Abonnements und eine benutzerdefinierte zustandsbehaftete Transformation, die Datensätze hält, bis sie sie einander zuordnen kann.

Voraussetzungen

Erfüllen Sie die gemeinsamen Beispielvoraussetzungen, einschließlich des bestehenden default Profils, factory-mqtt Endpunkts und factory-artifacts Registry-Endpunkts für ghcr.io. Diese Beispiele beziehen sich auf diese gemeinsamen Ressourcen, ohne sie nachzubilden.

Für Azure CLI und Bicep folgen Sie den Anweisungen zur Einrichtung der Umgebung in Datenflussdiagramm erstellen. Legen Sie AIO_INSTANCE_NAME und RESOURCE_GROUP auf deine bestehende Instanz und Ressourcengruppe fest. Für Bicep sollten Sie auch CUSTOM_LOCATION_NAME auf den bestehenden benutzerdefinierten Standortnamen der Instanz einstellen.

Wählen Sie die Matching-Regel

Zeitstempel und Asset-ID: eine Geschwindigkeitsmessung mit der Temperatur abgleichen, die zum Zeitpunkt der Erfassung vorlag.
Element- und Asset-IDs: Abgleich einer Vorhersage mit einer verzögerten Messung desselben physischen Teils.

Die benutzerdefinierte Transformation verwaltet ausstehende Datensätze, Ablaufdatum und Duplikatbehandlung. Führen Sie jeden Samplegraphen auf einer Laufzeitinstanz aus: Ihr Zustand befindet sich im Speicher und geht beim Neustart verloren. Die Wiederherstellung erfordert externe, dauerhafte Input-Logs und nachgelagerte Deduplizierung.

Kontext nach Zeitstempel abgleichen

Ein Geschwindigkeitswert von 12 bei 10:00:02 entspricht einer Temperatur von 20 bei 10:00:01, nicht der Temperatur von 22 bei 10:00:03. Die spätere Temperatur ist beim Zeitstempel der Geschwindigkeitsmessung nicht gültig.

Diagramm einer MQTT-Quelle, die Geschwindigkeit, Temperatur und Tick-Nachrichten an einen benutzerdefinierten Zeitstempel-Matcher überträgt. Er gleicht Geschwindigkeit 12 mit der früheren Temperatur 20 ab und veröffentlicht einen ausgerichteten Datensatz.

Wählen Sie den neuesten Kontext zum Zeitpunkt des Zeitstempels der Lesung oder davor. Diese Auswahl erfordert benutzerdefiniertes stateful WASM oder einen externen Dienst, kein eingebautes Fenster.

Diese Verkabelung verwendet das benutzerdefinierte graph-align-by-time-Beispiel im factory-artifacts-Registry-Endpunkt für ghcr.io. Die Beispielquelle und die Verträge beschreiben ihren gebundenen Zustand und die Diagnose. Der Vertrag für dieses Beispiel lautet: Buffer nach Asset; zwei Sekunden für Kontext in falscher Reihenfolge zulassen; die neueste Temperatur am oder vor dem Geschwindigkeitszeitstempel auswählen, höchstens fünf Sekunden alt; die Messung senden, wenn diese Entscheidung bereit ist. Wenn der relevante Kontext fehlt, gib eine Diagnosemeldung aus und überspringe den Lesevorgang.

Die Vorgangsanzeige unterstützt nur das Erstellen und Anzeigen von Diagrammartefakten aus der Azure Container Registry (ACR) und für integrierte Transformationen, mcr.microsoft.com. Es kann das GHCR-Artefakt factory-artifactsdieser Probe nicht auswählen oder anzeigen. Verwenden Sie die Azure CLI- oder Bicep-Registerkarte, um das exakte Beispiel bereitzustellen.

Für eine separat neu veröffentlichte ACR-Variante mit einem konfigurierten ACR-Registry-Endpunkt:

Erstellen Sie einen Datenflussgraphen namens default mit dem bestehenden align-by-time-Profil.
Konfigurieren Sie eine Quelle mit factory-mqtt und Topics factory/speed und factory/temperature.
Wählen Sie Graphtransformation hinzufügen (optional). In der Graphenauswahl wählen Sie das neu veröffentlichte graph-align-by-time:1.0.0-rc.4 Artefakt aus und wählen Sie Anwenden.
Konfigurieren Sie das Ziel mit factory-mqtt und dem Thema factory/aligned, dann wählen Sie Speichern. Die Vorgangsanzeige erzeugt automatisch die Knotenverbindungen des Artefakts.

Diese ACR-Variante verwendet einen anderen Registry-Endpunkt; es ist nicht die exakte GHCR-Konfiguration in den folgenden Bereitstellungs-Tabs. Für unterstützte Aktionen und Einschränkungen siehe WASM-Transformationen verwenden.

Speichern Sie diese vollständigen Diagrammeigenschaften als graph.json:

{
  "mode": "Enabled",
  "nodes": [
    {
      "name": "source",
      "nodeType": "Source",
      "sourceSettings": {
        "endpointRef": "factory-mqtt",
        "dataSources": ["factory/speed", "factory/temperature"]
      }
    },
    {
      "name": "processor",
      "nodeType": "Graph",
      "graphSettings": {
        "registryEndpointRef": "factory-artifacts",
        "artifact": "azure-samples/explore-iot-operations/dataflow-patterns/graph-align-by-time:1.0.0-rc.4"
      }
    },
    {
      "name": "destination",
      "nodeType": "Destination",
      "destinationSettings": {
        "endpointRef": "factory-mqtt",
        "dataDestination": "factory/aligned"
      }
    }
  ],
  "nodeConnections": [
    {"from": {"name": "source"}, "to": {"name": "processor"}},
    {"from": {"name": "processor"}, "to": {"name": "destination"}}
  ]
}

Diagramm anwenden:

az iot ops dataflowgraph apply --name align-by-time --instance $AIO_INSTANCE_NAME --resource-group $RESOURCE_GROUP --config-file graph.json

Speichere diese vollständige Vorlage als main.bicep. Es verweist auf die bestehende Instanz, den benutzerdefinierten Standort und das Profil:

param aioInstanceName string
param customLocationName string

resource aioInstance 'Microsoft.IoTOperations/instances@2026-07-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-07-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  parent: defaultDataflowProfile
  name: 'align-by-time'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        name: 'source'
        nodeType: 'Source'
        sourceSettings: {
          endpointRef: 'factory-mqtt'
          dataSources: [
            'factory/speed'
            'factory/temperature'
          ]
        }
      }
      {
        name: 'processor'
        nodeType: 'Graph'
        graphSettings: {
          registryEndpointRef: 'factory-artifacts'
          artifact: 'azure-samples/explore-iot-operations/dataflow-patterns/graph-align-by-time:1.0.0-rc.4'
        }
      }
      {
        name: 'destination'
        nodeType: 'Destination'
        destinationSettings: {
          endpointRef: 'factory-mqtt'
          dataDestination: 'factory/aligned'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'source' }
        to: { name: 'processor' }
      }
      {
        from: { name: 'processor' }
        to: { name: 'destination' }
      }
    ]
  }
}

Stellen Sie die Vorlage bereit:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep --parameters aioInstanceName="$AIO_INSTANCE_NAME" customLocationName="$CUSTOM_LOCATION_NAME"

Important

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

Speichere diese vollständige Ressource als align-by-time.yaml:

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: align-by-time
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  nodes:
    - name: source
      nodeType: Source
      sourceSettings:
        endpointRef: factory-mqtt
        dataSources: [factory/speed, factory/temperature]
    - name: processor
      nodeType: Graph
      graphSettings:
        registryEndpointRef: factory-artifacts
        artifact: azure-samples/explore-iot-operations/dataflow-patterns/graph-align-by-time:1.0.0-rc.4
    - name: destination
      nodeType: Destination
      destinationSettings:
        endpointRef: factory-mqtt
        dataDestination: factory/aligned
  nodeConnections:
    - from: { name: source }
      to: { name: processor }
    - from: { name: processor }
      to: { name: destination }

Diagramm zum Debuggen anwenden:

kubectl apply -f align-by-time.yaml

Veröffentlichen Sie diese synthetischen Nachrichten in der Reihenfolge, eine JSON-Nachricht pro Zeile (NDJSON):

{"kind":"temperature","assetId":"line-1","sampledAt":"2026-10-19T10:00:01Z","temperature":20}
{"kind":"temperature","assetId":"line-1","sampledAt":"2026-10-19T10:00:03Z","temperature":22}
{"kind":"speed","assetId":"line-1","sampledAt":"2026-10-19T10:00:02Z","speed":12}
{"kind":"tick","assetId":"line-1","sampledAt":"2026-10-19T10:00:04Z"}

Die ersten beiden Nachrichten werden an factory/temperature gesendet; der Geschwindigkeitswert und der abschließende Tick werden an factory/speed gesendet. Der Transformation behält die Temperaturen bei und hält den Geschwindigkeitswert. Der Tick bei 10:00:04 gibt den Geschwindigkeitswert bei 10:00:02 nach der Zwei-Sekunden-Frist frei, was dieses Ergebnis auf factory/aligned liefert:

{
  "assetId": "line-1",
  "sampledAt": "2026-10-19T10:00:02Z",
  "speed": 12,
  "temperature": 20,
  "temperatureSampledAt": "2026-10-19T10:00:01Z"
}

Wenn ein Eingang mehrere Messwerte ausgibt, gibt das benutzerdefinierte Modul {"kind":"batch","records":[...]} aus. Die nachgelagerte Verarbeitung muss diesen Stapelumschlag verarbeiten.

Behalten Sie diese Laufzeitbeschränkungen im Hinterkopf:

Senden Sie vertrauenswürdige periodische Ticks für jedes aktive Asset. Es gibt keinen Hintergrundtimer, daher können Leerlauf-Graphen keine ausstehenden Werte veröffentlichen oder den Zustand ohne einen weiteren Eingang ablaufen lassen.
Verwenden Sie RFC3339 Zeitstempel, nicht früher als die Unix-Epoche, mit Millisekundenpräzision oder gröber. Die standardmäßige maximale Uhrabweichung in der Zukunft beträgt 30 Sekunden.
Überwachen Sie Diagnosedaten auf verspätete oder doppelte Eingaben und fehlenden Kontext. Um die Grenzen des Beispiels und die Duplizierungsregeln zu sehen, überprüfen Sie die Datei contracts.json.
Behalte gepufferte Zeitstempel und IDs in JSON. Ausgabe-MQTT-Metadaten stammen vom Eingang, der das Ergebnis freigibt, nicht unbedingt von der ursprünglichen Messung.

Vorhersagen mit verzögerten Messungen per ID abgleichen

Gleiche die Vorhersage 203 mit der späteren Messung 200 für das Asset line-1, Stück part-42. Das Residuum ist die Messung minus Vorhersage: 200 - 203 = -3.

Diagramm einer MQTT-Quelle, die Vorhersagen und Messungen an einen benutzerdefinierten ID-Matcher weiterleitet. Es stimmt line-1 und part-42 ab und veröffentlicht residual minus 3 sowie TargetError 2.

Abgleich nach Asset- und Teil-IDs. Füge eine Durchlauf- oder Batch-ID hinzu, falls sich die Stück-IDs wiederholen. Speichern Sie ausstehende Datensätze für beide Ankunftsreihenfolgen, begrenzen Sie deren Alter und Anzahl und bewahren Sie die Zeitstempel sowie die Modellversion der Vorhersage auf.

Das benutzerdefinierte graph-correlate-by-id-Beispiel gleicht Datensätze ab und berechnet residual und targetError. Ein Residual misst den Vorhersagefehler; bei einem Ziel von 198ist der separate Zielfehler 200 - 198 = 2.

Stellen Sie das ID-Korrelationsbeispiel bereit

Die Vorgangsanzeige unterstützt nur das Erstellen und Anzeigen von Diagrammartefakten aus ACR und für integrierte Transformationen, mcr.microsoft.com. Es kann das GHCR-Artefakt factory-artifactsdieser Probe nicht auswählen oder anzeigen. Verwenden Sie die Azure CLI- oder Bicep-Registerkarte, um das exakte Beispiel bereitzustellen.

Für eine separat neu veröffentlichte ACR-Variante mit einem konfigurierten ACR-Registry-Endpunkt:

Erstellen Sie einen Datenflussgraphen namens default mit dem bestehenden correlate-predictions-Profil.
Konfigurieren Sie eine Quelle mit factory-mqtt und Topics factory/predictions und factory/measurements.
Wählen Sie Graphtransformation hinzufügen (optional). In der Graphenauswahl wählen Sie das neu veröffentlichte graph-correlate-by-id:1.0.0-rc.4 Artefakt aus und wählen Sie Anwenden.
Konfigurieren Sie das Ziel mit factory-mqtt und dem Thema factory/matches, dann wählen Sie Speichern. Die Vorgangsanzeige erzeugt automatisch die Knotenverbindungen des Artefakts.

Speichern Sie diese vollständigen Diagrammeigenschaften als graph.json:

{
  "mode": "Enabled",
  "nodes": [
    {
      "name": "source",
      "nodeType": "Source",
      "sourceSettings": {
        "endpointRef": "factory-mqtt",
        "dataSources": ["factory/predictions", "factory/measurements"]
      }
    },
    {
      "name": "processor",
      "nodeType": "Graph",
      "graphSettings": {
        "registryEndpointRef": "factory-artifacts",
        "artifact": "azure-samples/explore-iot-operations/dataflow-patterns/graph-correlate-by-id:1.0.0-rc.4"
      }
    },
    {
      "name": "destination",
      "nodeType": "Destination",
      "destinationSettings": {
        "endpointRef": "factory-mqtt",
        "dataDestination": "factory/matches"
      }
    }
  ],
  "nodeConnections": [
    {"from": {"name": "source"}, "to": {"name": "processor"}},
    {"from": {"name": "processor"}, "to": {"name": "destination"}}
  ]
}

Diagramm anwenden:

az iot ops dataflowgraph apply --name correlate-predictions --instance $AIO_INSTANCE_NAME --resource-group $RESOURCE_GROUP --config-file graph.json

Speichere diese vollständige Vorlage als main.bicep. Es verweist auf die bestehende Instanz, den benutzerdefinierten Standort und das Profil:

param aioInstanceName string
param customLocationName string

resource aioInstance 'Microsoft.IoTOperations/instances@2026-07-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-07-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-07-01' = {
  parent: defaultDataflowProfile
  name: 'correlate-predictions'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        name: 'source'
        nodeType: 'Source'
        sourceSettings: {
          endpointRef: 'factory-mqtt'
          dataSources: [
            'factory/predictions'
            'factory/measurements'
          ]
        }
      }
      {
        name: 'processor'
        nodeType: 'Graph'
        graphSettings: {
          registryEndpointRef: 'factory-artifacts'
          artifact: 'azure-samples/explore-iot-operations/dataflow-patterns/graph-correlate-by-id:1.0.0-rc.4'
        }
      }
      {
        name: 'destination'
        nodeType: 'Destination'
        destinationSettings: {
          endpointRef: 'factory-mqtt'
          dataDestination: 'factory/matches'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'source' }
        to: { name: 'processor' }
      }
      {
        from: { name: 'processor' }
        to: { name: 'destination' }
      }
    ]
  }
}

Stellen Sie die Vorlage bereit:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep --parameters aioInstanceName="$AIO_INSTANCE_NAME" customLocationName="$CUSTOM_LOCATION_NAME"

Important

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

Azure Samples bietet eine vollständige correlate-predictions.yaml mit eigenen Themen und veröffentlichter Graphenreferenz. Die separate align-by-time.yaml enthält das Zeitstempelbeispiel.

Klone das Sample-Repository und führe diese Befehle aus, nachdem du die gemeinsamen Voraussetzungen erfüllt hast:

git clone https://github.com/Azure-Samples/explore-iot-operations.git
cd explore-iot-operations
kubectl apply -f samples/wasm/dataflow-patterns/manifests/registry.yaml
kubectl apply -f samples/wasm/dataflow-patterns/manifests/correlate-predictions.yaml

Diese Manifeste verwenden das bestehende default-Profil und den factory-mqtt-Endpunkt in azure-iot-operations. Das Registry-Manifest beschreibt den gemeinsamen factory-artifacts Endpunkt; wenn er bereits konfiguriert ist, ist sein Apply-Schritt idempotent.

ID-Korrelation erfordert Vorhersagen mit kind: "prediction", assetId, pieceId, einer positiven Ganzzahl modelVersion und endlichen numerischen predictedWeight. Messungen erfordern kind: "measurement", dieselben Asset- und Stück-IDs sowie endliche numerische measuredWeight und targetWeight. Verwenden Sie verschiedene predictedAt und measuredAt Felder für optionale Zeitstempel, die verschiedene Momente beschreiben. Gemeinsame, nicht-schlüsselbasierte Felder müssen gleiche Werte haben, sonst schlägt das Match fehl.

Das benutzerdefinierte Modul verwaltet ausstehende Matches, Limits, Duplikatbearbeitung und abgeschlossene Match-Ausgaben. Der Graph veröffentlicht Übereinstimmungen zum Thema factory/matches.

Synthetische Eingabenachrichten (NDJSON): Veröffentlichen Sie die erste Zeile an factory/predictions und die zweite Zeile später in factory/measurements. Die Vorhersage enthält die ursprünglichen Merkmalswerte für das Nachtrainieren.

{"kind":"prediction","assetId":"line-1","pieceId":"part-42","modelVersion":8,"speed":10,"temperature":20,"predictedWeight":203}
{"kind":"measurement","assetId":"line-1","pieceId":"part-42","measuredWeight":200,"targetWeight":198}

Die Vorhersage allein liefert keine Übereinstimmung. Wenn die Messung eintrifft, erwarten Sie diese Payload auf factory/matches, einschließlich der ursprünglichen Modellversion und der Funktionswerte:

{"kind":"correlated","assetId":"line-1","pieceId":"part-42","modelVersion":8,"speed":10,"temperature":20,"predictedWeight":203,"measuredWeight":200,"targetWeight":198,"residual":-3,"targetError":2}

Jede der beiden Ankunftsreihenfolgen funktioniert. Standardmäßig laufen ausstehende Einträge 60 Sekunden nach dem Eintreffen der ersten Seite ab; weitere Versuche verlängern diese Zeit nicht. Gültige Eingänge oder vertrauenswürdige {"kind":"tick","assetId":"line-1"}-Nachrichten lösen die Bereinigung aus; es gibt keinen Hintergrundtimer.

Abgeschlossene Matches werden standardmäßig für 60 Sekunden dedupliziert, nicht für immer. Bei einem Neustart gehen sowohl ausstehende als auch abgeschlossene Einträge verloren. Zur Wiederherstellung deduplizieren Sie in einer externen dauerhaften Senke nach Anlage, Stück und optionaler Ausführungs-ID. Zum erneuten Trainieren bewahren Sie die ursprünglichen Featurewerte auf, die von jeder Vorhersage verwendet werden.

Feedback

War diese Seite hilfreich?

Zuletzt aktualisiert um 2026-10-30