Justera och korrelera poster

Para ihop relaterade värden som kommer som separata meddelanden. Dessa exempel använder en MQTT-källa med flera prenumerationer och en anpassad stateful transform som håller poster tills den kan matcha dem.

Förutsättningar

Slutför de förutsättningarna för det delade exemplet, inklusive den befintliga default profilen, factory-mqtt endpointen och factory-artifacts registerendpointen för ghcr.io. Dessa exempel refererar till dessa gemensamma resurser utan att återskapa dem.

För Azure CLI och Bicep, följ miljöuppsättningen i Skapa ett dataflödesdiagram. Sätt AIO_INSTANCE_NAME och RESOURCE_GROUP till din befintliga instans respektive resursgrupp. För Bicep, sätt CUSTOM_LOCATION_NAME även till instansens befintliga anpassade platsnamn.

Välj matchningsregeln

Tidsstämpel och resurs-ID: matcha en hastighetsavläsning mot temperaturen som gällde vid provtagningen.
Artikel- och tillgångs-ID: Matcha en prognos med en fördröjd mätning av samma fysiska objekt.

Den anpassade transformen hanterar väntande poster, utgång och dubbletthantering. Kör varje samplegraf på en runtime-instans: dess tillstånd finns i minnet och försvinner vid omstart. Återhämtning kräver externa, hållbara indataloggar och nedströms deduplicering.

Matcha kontext utifrån tidsstämpel

En snabbavläsning av 12 vid 10:00:02 matchar temperaturen 20 vid 10:00:01, inte temperaturen 22 vid 10:00:03. Den senare temperaturen är inte giltig vid hastighetsavläsningens tidsstämpel.

Diagram över en MQTT-källa som vidarebefordrar hastighet, temperatur och tick-meddelanden till en anpassad tidsstämpelmatchare. Den matchar hastighet 12 med den tidigare temperaturen 20 och publicerar en justerad post.

Välj den senaste kontexten vid eller före läsningens tidsstämpel. Detta val kräver anpassad stateful WASM eller en extern tjänst, inte ett inbyggt fönster.

Denna konfiguration använder det anpassade graph-align-by-time exemplet i registerändpunkten factory-artifacts för ghcr.io. Exempelkällan och kontrakten beskriver dess avgränsade tillstånd och diagnostik. Kontraktet för detta exempel är: buffert per asset; tillåt två sekunder för kontext i fel ordning; välj den senaste temperaturen vid eller före hastighetstidsstämpeln, högst fem sekunder gammal; skicka ut avläsningen när beslutet är klart. Om giltig kontext saknas, skicka ett diagnostikmeddelande och hoppa över avläsningen.

Upplevelsen för åtgärder stöder endast skapande och visning av grafartefakter från Azure Container Registry (ACR) och, för inbyggda transformationer, mcr.microsoft.com. Den kan inte välja eller visa denna exempels GHCR-artefakt genom factory-artifacts. Använd fliken Azure CLI eller Bicep för att distribuera det exakta exemplet.

För en separat ompublicerad ACR-variant med en konfigurerad ACR-registerslutpunkt:

Skapa en dataflödesgraf som är namngiven align-by-time med den befintliga default profilen.
Konfigurera en källa med factory-mqtt och ämnena factory/speed och factory/temperature.
Välj Lägg till graftransformering (valfritt). Vid Grafval, välj den återpublicerade graph-align-by-time:1.0.0-rc.4 artefakten och välj Tillämpa.
Konfigurera destinationen med factory-mqtt och ämnet factory/aligned, välj sedan Spara. Åtgärdsupplevelsen genererar automatiskt artefaktens nodanslutningar.

Denna ACR-variant använder en annan registerslutpunkt; det är inte exakt GHCR-konfigurationen i följande distributionsflikar. För stödda åtgärder och begränsningar, se Använd WASM-transformer.

Spara dessa kompletta grafegenskaper som 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"}}
  ]
}

Använd grafen:

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

Spara denna kompletta mall som main.bicep. Den refererar till den befintliga instansen, den anpassade platsen och profilen:

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' }
      }
    ]
  }
}

Distribuera mallen:

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

Important

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

Spara hela denna resurs som 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 }

Använd grafen för felsökning:

kubectl apply -f align-by-time.yaml

Publicera dessa syntetiska meddelanden i ordning, ett JSON-meddelande per rad (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"}

De två första meddelandena går till factory/temperature; speed reading och sista ticken går till factory/speed. transformen behåller temperaturerna och behåller hastighetsvärdet. Markeringen vid 10:00:04 aktiverar hastighetsavläsningen vid 10:00:02 efter två sekunders tillåten tid, vilket ger detta resultat på factory/aligned:

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

När en ankomst släpper flera avläsningar sänder den anpassade modulen {"kind":"batch","records":[...]}. Nedströmsprocessering måste hantera batchkuvertet.

Ha dessa körtidsbegränsningar i åtanke:

Skicka betrodda periodiska tickdata för varje aktiv tillgång. Det finns ingen bakgrundstimer, så inaktiva grafer kan inte släppa väntande avläsningar eller låta tillstånd upphöra att gälla utan en ny ankomst.
Använd RFC3339 tidsstämplar, inte tidigare än Unix-epoken, med millisekundprecision eller grövare. Standardvärdet för maximal framtida klockskevhet är 30 sekunder.
Övervaka diagnostik för sena eller duplicerade inmatningar och saknad kontext. För att se exemplets gränser och dubblettregler, kontrollera filen contracts.json.
Behåll buffrade tidsstämplar och ID:n i JSON. Utgående MQTT-metadata kommer från händelsen som genererar resultatet, inte nödvändigtvis från den ursprungliga avläsningen.

Matcha prediktioner med fördröjda mätningar via ID

Matcha prediktion 203 med senare mätning 200 för asset line-1, del part-42. Residualen är mätning minus prediktion: 200 - 203 = -3.

Diagram över en MQTT-käll som vidarebefordrar prediktioner och mätningar till en anpassad ID-matchare. Den matchar rad 1 och del 42 och publicerar residual minus 3 och targetError 2.

Matcha utifrån resurs- och del-ID. Inkludera ett kör- eller batch-ID om stycke-ID:n upprepas. Håll väntande poster oavsett ankomstordning, begränsa deras ålder och antal samt bevara förutsägelsens tidsstämplar och modellversion.

Det anpassade graph-correlate-by-id exemplet matchar poster och beräknar residual och targetError. Ett residual mäter prediktionsfel; med ett mål på 198 är det separata målfelet 200 - 198 = 2.

Distribuera exemplet med ID-korrelation

Upplevelsen för åtgärder stöder endast skapande och visning av grafartefakter från ACR och, för inbyggda transformationer, mcr.microsoft.com. Den kan inte välja eller visa denna exempels GHCR-artefakt genom factory-artifacts. Använd fliken Azure CLI eller Bicep för att distribuera det exakta exemplet.

För en separat ompublicerad ACR-variant med en konfigurerad ACR-registerslutpunkt:

Skapa en dataflödesgraf som är namngiven correlate-predictions med den befintliga default profilen.
Konfigurera en källa med factory-mqtt och ämnena factory/predictions och factory/measurements.
Välj Lägg till graftransformering (valfritt). Vid Grafval, välj den återpublicerade graph-correlate-by-id:1.0.0-rc.4 artefakten och välj Tillämpa.
Konfigurera destinationen med factory-mqtt och ämnet factory/matches, välj sedan Spara. Åtgärdsupplevelsen genererar automatiskt artefaktens nodanslutningar.

Spara dessa kompletta grafegenskaper som 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"}}
  ]
}

Använd grafen:

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

Spara denna kompletta mall som main.bicep. Den refererar till den befintliga instansen, den anpassade platsen och profilen:

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' }
      }
    ]
  }
}

Distribuera mallen:

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

Important

Användning av Kubernetes-distributionsmanifest stöds inte i produktionsmiljöer och bör endast användas för felsökning och testning.

Azure Samples tillhandahåller en komplett correlate-predictions.yaml med egna ämnen och publicerad grafreferens. Den separata align-by-time.yaml innehåller exemplet på tidsstämpel.

Klona sample-repositoryt och kör dessa kommandon efter att ha slutfört de gemensamma förutsättningarna:

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

Dessa manifest använder den befintliga default profilen och factory-mqtt slutpunkten i azure-iot-operations. Registermanifestet beskriver den delade factory-artifacts ändpunkten; om den redan är konfigurerad är dess appliceringssteg idempotent.

ID-korrelation kräver prediktioner med modelVersion, predictedWeight, pieceId, ett positivt heltal assetId, och ändligt numeriskt kind: "prediction". Mätningar kräver kind: "measurement", samma tillgångs- och stycke-ID:n samt ändliga numeriska measuredWeight och targetWeight. Använd olika predictedAt fält measuredAt för valfria tidsstämplar som beskriver olika ögonblick. Delade icke-nyckelfält måste ha lika värden, annars misslyckas matchningen.

Den anpassade modulen hanterar väntande matchningar, gränser, hantering av dubbletter och utdata för slutförda matchningar. Grafen publicerar matchningar till det separata ämnet factory/matches.

Syntetiska inmatningsmeddelanden (NDJSON): publicera den första raden till factory/predictions och den andra raden senare till factory/measurements. Prediktionen inkluderar de ursprungliga funktionsvärdena för omträning.

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

Prediktionen i sig ger ingen match. När mätningen anländer, förvänta dig denna nyttolast på factory/matches, inklusive den ursprungliga modellversionen och funktionsvärdena:

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

Vilken ankomstordning som helst fungerar. Som standard upphör väntande poster 60 sekunder efter att den första parten anländer; omförsök förlänger inte den tiden. Giltiga inkommande meddelanden eller betrodda {"kind":"tick","assetId":"line-1"} meddelanden utlöser rensning; det finns ingen bakgrundstimer.

Slutförda matcher är deduplicerade i 60 sekunder som standard, inte för alltid. Vid omstart förloras både väntande och slutförda poster. För återställning, deduplicera i ett externt hållbart lagringsmål efter resurs, del och valfritt run ID. För omträning, bevara de ursprungliga funktionsvärdena som används av varje prediktion.

Feedback

Var den här sidan till hjälp?

Uppdaterades senast 2026-10-29