Aligner et corréler les enregistrements

Faites correspondre les valeurs associées qui arrivent sous forme de messages séparés. Ces exemples utilisent une source MQTT avec plusieurs abonnements et une transformation étatique personnalisée qui conserve les enregistrements jusqu’à ce qu’elle puisse les faire correspondre.

Prerequisites

Compléter les prérequis d’exemples partagés, y compris le profil existant default , factory-mqtt le point de terminaison et factory-artifacts le point de terminaison du registre pour ghcr.io. Ces exemples font référence à ces ressources partagées sans les recréer.

Pour Azure CLI et Bicep, suivez la configuration de l’environnement dans Create a data flow graph. Définissez AIO_INSTANCE_NAME et RESOURCE_GROUP sur votre instance existante et votre groupe de ressources existant. Pour Bicep, définissez également CUSTOM_LOCATION_NAME sur le nom d’emplacement personnalisé existant de l’instance.

Choisissez la règle de correspondance

Horodatage et identifiant de l’actif : faire correspondre une lecture de vitesse à la température en vigueur au moment où elle a été échantillonnée.
IDs d’éléments et de ressources : associer une prédiction à une mesure différée de la même pièce physique.

La transformation personnalisée gère les enregistrements en attente, les expirations et la gestion des doublons. Exécutez chaque exemple de graphe sur une instance d’exécution : son état est en mémoire et est perdu au redémarrage. La reprise nécessite des journaux d’entrée externes durables et une déduplication en aval.

Contexte de correspondance par horodatage

Une mesure de vitesse de 12 en 10:00:02 correspond à la température 20 à 10:00:01, et non à la température 22 à 10:00:03. La température ultérieure n’est pas valide à l’horodatage du relevé de vitesse.

Diagramme d’une source MQTT qui transmet les messages de vitesse, température et tick vers un moteur de correspondance d’horodatage personnalisé. Il fait correspondre la vitesse 12 avec la température 20 antérieure et publie un enregistrement aligné.

Sélectionnez le dernier contexte à l’horodatage de la lecture ou avant. Cette sélection nécessite un WASM avec état personnalisé ou un service externe, pas une fenêtre intégrée.

Cette configuration utilise l’échantillon personnalisé graph-align-by-time dans le endpoint du registre factory-artifacts pour ghcr.io. Le code source d’exemple et les contrats décrivent son état borné et ses diagnostics. Le contrat pour cet exemple est : tampon par ressource ; prévoir deux secondes pour un contexte reçu dans le désordre ; sélectionner la température la plus récente à l’horodatage de vitesse ou avant celui-ci, âgée d’au plus cinq secondes ; émettre la lecture lorsque cette décision est prête. Si un contexte éligible manque, émettez un diagnostic et ignorez la lecture.

L’expérience opérationnelle ne supporte que la création et la visualisation d’artefacts de graphes depuis Azure Container Registry (ACR) et, pour les transformations intégrées, mcr.microsoft.com. Il ne peut pas sélectionner ni afficher l’artefact GHCR de cet échantillon via factory-artifacts. Utilisez l’onglet Azure CLI ou Bicep pour déployer l’échantillon exact.

Pour une variante ACR republiée séparément avec un point de terminaison de registre ACR configuré :

Créez un graphe de flux de données nommé align-by-time avec le profil existant default .
Configurez une source en utilisant factory-mqtt et les rubriques factory/speed et factory/temperature.
Sélectionnez Ajouter une transformation de graphique (facultatif). Dans la sélection Graph, sélectionnez l’artefact republié graph-align-by-time:1.0.0-rc.4 et sélectionnez Appliquer.
Configurez la destination à l’aide de factory-mqtt et de la rubrique factory/aligned, puis sélectionnez Enregistrer. L’expérience Operations génère automatiquement les connexions de nœud de l’artefact.

Cette variante ACR utilise un point de terminaison de registre différent ; ce n’est pas exactement la configuration GHCR dans les onglets de déploiement suivants. Pour les actions prises en charge et les limitations, voir Utiliser les transformations WASM.

Enregistrez ces propriétés du graphe complet sous la forme 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"}}
  ]
}

Appliquer le graphique :

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

Enregistrez ce modèle complet sous la forme de main.bicep. Il fait référence à l’instance existante, à son emplacement personnalisé et au 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' }
      }
    ]
  }
}

Déployez le modèle :

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

Important

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

Enregistrez cette ressource complète sous la forme 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 }

Utiliser le graphique pour le débogage :

kubectl apply -f align-by-time.yaml

Publiez ces messages synthétiques dans l’ordre, un message JSON par ligne (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"}

Les deux premiers messages vont à factory/temperature; la lecture rapide et le tick final vont à factory/speed. La transformation conserve les températures et maintient la valeur de vitesse. La coche à 10:00:04 déclenche la lecture de vitesse à 10:00:02 après le délai de deux secondes, produisant ce résultat sur factory/aligned:

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

Lorsqu’une arrivée libère plusieurs lectures, le module personnalisé émet {"kind":"batch","records":[...]}. Le traitement en aval doit gérer cette enveloppe de lot.

Gardez à l’esprit ces contraintes d’exécution :

Envoyez des ticks périodiques fiables pour chaque actif en activité. Il n’y a pas de minuterie en arrière-plan, donc les graphiques d’inactivité ne peuvent pas libérer les relevés en attente ni expirer l’état sans nouvelle arrivée.
Utilisez des horodatages RFC3339, pas antérieurs à l’époque Unix, avec une précision milliseconde ou plus grossière. Le décalage maximal futur par défaut de l’horloge est de 30 secondes.
Surveillez les diagnostics pour détecter des entrées tardives ou dupliquées et le contexte manquant. Pour voir les limites et les règles de duplication de l’échantillon, vérifiez le contracts.json fichier.
Garde les horodatages et identifiants mis en tampon en JSON. Les métadonnées MQTT de sortie proviennent de l’arrivée qui émet le résultat, pas nécessairement de la lecture initiale.

Faire correspondre les prédictions aux mesures différées par ID

Faire correspondre la prédiction 203 avec la mesure ultérieure 200 pour l’actif line-1, pièce part-42. Le résidu est la mesure moins la prédiction : 200 - 203 = -3.

Diagramme d’une source MQTT transférant les prédictions et mesures vers un composant de correspondance d’ID personnalisé. Il met en correspondance line-1 et part-42 et publie le résiduel moins 3 et targetError 2.

Faire correspondre à l’aide des ID de ressource et de pièce. Incluez un identifiant de série ou de lot si les identifiants de pièces se répètent. Conservez les dossiers en attente pour chaque ordre d’arrivée, limitez leur ancienneté et leur nombre, et conservez les horodatages et la version du modèle de la prédiction.

L’échantillon personnalisé graph-correlate-by-id met en correspondance les enregistrements et calcule residual et targetError. Un résidu mesure l’erreur de prédiction ; avec une cible de 198, l’erreur cible distincte est 200 - 198 = 2.

Déployez l’exemple de corrélation d’ID

L’expérience opérationnelle ne supporte que la création et la visualisation d’artefacts de graphes à partir d’ACR et, pour les transformations intégrées, mcr.microsoft.com. Il ne peut pas sélectionner ni afficher l’artefact GHCR de cet échantillon via factory-artifacts. Utilisez l’onglet Azure CLI ou Bicep pour déployer l’échantillon exact.

Pour une variante ACR republiée séparément avec un point de terminaison de registre ACR configuré :

Créez un graphe de flux de données nommé correlate-predictions avec le profil existant default .
Configurez une source en utilisant factory-mqtt et les rubriques factory/predictions et factory/measurements.
Sélectionnez Ajouter une transformation de graphique (facultatif). Dans la sélection Graph, sélectionnez l’artefact republié graph-correlate-by-id:1.0.0-rc.4 et sélectionnez Appliquer.
Configurez la destination à l’aide de factory-mqtt et de la rubrique factory/matches, puis sélectionnez Enregistrer. L’expérience Operations génère automatiquement les connexions de nœud de l’artefact.

Enregistrez ces propriétés du graphe complet sous la forme 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"}}
  ]
}

Appliquer le graphique :

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

Enregistrez ce modèle complet sous la forme de main.bicep. Il fait référence à l’instance existante, à son emplacement personnalisé et au 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' }
      }
    ]
  }
}

Déployez le modèle :

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

Important

L’utilisation des manifestes de déploiement Kubernetes n’est pas prise en charge dans les environnements de production et ne doit être utilisée que pour le débogage et le test.

Azure Samples fournit un complet correlate-predictions.yaml avec ses propres sujets et sa référence au graphe publié. Le séparé align-by-time.yaml contient l’exemple de l’horodatage.

Clonez le dépôt d’échantillons et exécutez ces commandes après avoir rempli les prérequis partagés :

git clone --branch huguesBouvier/dataflow-pattern-samples 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

Ces manifestes utilisent le profil existant default et le point de terminaison factory-mqtt dans azure-iot-operations. Le manifeste du registre décrit le point de terminaison partagé factory-artifacts ; s’il est déjà configuré, son étape d’application est idempotente.

La corrélation ID nécessite des prédictions avec kind: "prediction", assetId, pieceId, un entier modelVersionpositif , et un numérique predictedWeightfini . Les mesures nécessitent kind: "measurement", les mêmes identifiants d’actifs et de pièces, ainsi que des valeurs numériques finies measuredWeight et targetWeight. Utilisez des champs distincts predictedAt et measuredAt pour les horodatages optionnels qui décrivent différents instants. Les champs non clés partagés doivent avoir des valeurs égales, sinon la correspondance échoue.

Le module personnalisé gère les correspondances en attente, les limites, la gestion des doublons et la sortie des correspondances terminées. Le graphe publie les messages correspondants sur la rubrique distincte factory/matches.

Messages d’entrée synthétiques (NDJSON) : publier la première ligne vers factory/predictions et la seconde ligne plus tard vers factory/measurements. La prédiction inclut les valeurs originales des caractéristiques pour le réentraînement.

{"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}

La seule prédiction ne donne aucun résultat. Lorsque la mesure arrive, vous recevrez le payload suivant sur factory/matches, incluant la version originale du modèle et les valeurs des caractéristiques :

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

L’un ou l’autre ordre d’arrivée fonctionne. Par défaut, les entrées en attente expirent 60 secondes après l’arrivée de la première partie ; les réessais ne prolongent pas ce délai. Les arrivées valides ou les messages de confiance {"kind":"tick","assetId":"line-1"} déclenchent le nettoyage ; il n’y a pas de minuteur en arrière-plan.

Les matchs complétés sont dédupliqués pendant 60 secondes par défaut, pas pour toujours. Un redémarrage fait perdre à la fois les entrées en attente et les entrées terminées. Pour la récupération, dédupliquez dans un stockage persistant externe par asset, pièce et ID de run optionnel. Pour le réentraînement, conservez les valeurs d’origine des caractéristiques utilisées pour chaque prédiction.

Commentaires

Cette page a-t-elle été utile ?

Dernière mise à jour : 2026-10-29