Wyrównaj i skoreluj rekordy

Dopasowuj powiązane wartości, które pojawiają się jako osobne wiadomości. Te przykłady używają jednego źródła MQTT z wieloma subskrypcjami oraz niestandardowej transformacji stanowej, która przechowuje rekordy, dopóki nie będzie mogła dopasować odpowiadających im rekordów.

Prerequisites

Wypełnij wymagania wstępne wspólne dla przykładu, w tym istniejący default profil, factory-mqtt punkt końcowy i factory-artifacts punkt końcowy rejestru dla ghcr.io. Te przykłady odnoszą się do tych wspólnych zasobów bez ich odtwarzania.

W przypadku Azure CLI i Bicep postępuj zgodnie z konfiguracją środowiska w sekcji Utwórz wykres przepływu danych. Ustaw wartości AIO_INSTANCE_NAME i RESOURCE_GROUP dla istniejącego wystąpienia i grupy zasobów. Dla Bicep ustaw CUSTOM_LOCATION_NAME także na istniejącą, niestandardową nazwę lokalizacji instancji.

Wybierz regułę dopasowania

Znacznik czasu i identyfikator zasobów: dopasuj odczyt prędkości do temperatury obowiązującej w momencie pobierania próbki.
Identyfikatory przedmiotów i aktywów: dopasowanie prognozy do opóźnionego pomiaru tego samego fizycznego elementu.

Niestandardowa transformacja zarządza oczekującymi rekordami, wygaśnięciem i obsługą duplikatów. Uruchom każdy przykładowy graf na jednej instancji środowiska uruchomieniowego: jego stan znajduje się w pamięci i traci się po restarcie. Odzyskiwanie wymaga zewnętrznych, trwałych logów wejściowych oraz deduplikacji po stronie odbiorcy.

Dopasuj kontekst według znacznika czasu

Odczyt prędkości 12 przy 10:00:02 odpowiada temperaturze 20 przy 10:00:01, a nie temperaturze 22 przy 10:00:03. Późniejsza temperatura nie jest prawidłowa przy znaczniku czasu pomiaru prędkości.

Schemat pojedynczego źródła MQTT przesyłającego wiadomości o prędkości, temperaturze i tick do niestandardowego mechanizmu dopasowywania znaczników czasu. Dopasowuje prędkość 12 do wcześniejszej temperatury 20 i publikuje jeden wyrównany rekord.

Wybierz najnowszy kontekst w czasie odczytu lub przed nim. Ten wybór wymaga niestandardowego, stanowego WASM lub zewnętrznej usługi, a nie wbudowanego okna.

To okablowanie wykorzystuje niestandardową próbkę graph-align-by-time w punkcie końcowym rejestru factory-artifacts dla ghcr.io. Przykładowe źródło i kontrakty opisują jego ograniczony stan i diagnostykę. Założenia dla tego przykładu są następujące: buforuj według zasobu; uwzględnij dwie sekundy tolerancji dla danych przychodzących nie po kolei; wybierz najnowszą temperaturę w chwili znacznika czasu prędkości lub wcześniej, nie starszą niż pięć sekund; zwróć odczyt, gdy wynik tego wyboru jest gotowy. Jeśli brakuje odpowiedniego kontekstu, wygeneruj komunikat diagnostyczny i pomiń odczyt.

Środowisko operacyjne obsługuje jedynie tworzenie i wyświetlanie artefaktów grafu z usługi Azure Container Registry (ACR) oraz, w przypadku wbudowanych transformacji, mcr.microsoft.com. Nie może wybrać ani wyświetlić artefaktu GHCR tej próbki przez factory-artifacts. Użyj zakładki Azure CLI lub Bicep, aby wdrożyć dokładnie ten przykład.

Dla osobno ponownie opublikowanego wariantu ACR z skonfigurowanym punktem końcowym rejestru ACR:

Utwórz graf przepływu danych o nazwie align-by-time przy użyciu istniejącego profilu default.
Skonfiguruj źródło za pomocą factory-mqtt oraz tematów factory/speed i factory/temperature.
Wybierz Dodaj przekształcenie wykresu (opcjonalnie). W obszarze Wybór wykresu wybierz ponownie opublikowany graph-align-by-time:1.0.0-rc.4 artefakt i wybierz Zastosuj.
Skonfiguruj miejsce docelowe za pomocą factory-mqtt i tematufactory/aligned, a następnie wybierz Zapisz. Środowisko Operations Experience automatycznie generuje połączenia węzłów artefaktu.

Ten wariant ACR wykorzystuje inny punkt końcowy rejestru; nie jest to dokładna konfiguracja GHCR w poniższych zakładkach wdrożeń. W przypadku obsługiwanych akcji i ograniczeń zobacz Korzystanie z transformacji WASM.

Zapisz te właściwości pełnego grafu jako 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"}}
  ]
}

Zastosuj do wykresu:

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

Zapisz ten pełny szablon jako main.bicep. Odnosi się do istniejącej instancji, niestandardowej lokalizacji i profilu:

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

Wdróż szablon:

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

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

Zapisz ten kompletny materiał jako 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 }

Użyj wykresu do debugowania:

kubectl apply -f align-by-time.yaml

Publikuj te syntetyczne wiadomości po kolei, po jednej wiadomości JSON na linię (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"}

Pierwsze dwie wiadomości trafiają do factory/temperature; speed reading i ostatni tick idą do factory/speed. Transform zachowuje temperatury i zachowuje odczyt prędkości. Zaznaczenie przy 10:00:04 uruchamia odczyt prędkości przy 10:00:02 po upływie dwóch sekund, co daje ten wynik na factory/aligned:

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

Gdy jedno nadejście emituje kilka odczytów, moduł niestandardowy emituje {"kind":"batch","records":[...]}. Przetwarzanie w dalszej części musi obsłużyć ten batch envelope.

Miej na uwadze te ograniczenia środowiska uruchomieniowego:

Wysyłaj zaufane, okresowe ticki dla każdego aktywnego aktywa. Nie ma timera w tle, więc bezczynne grafy nie mogą zwolnić oczekujących odczytów ani wygasić stanu bez kolejnego przybycia.
Używaj znaczników czasu RFC3339, nie wcześniejszych niż epoka Uniksa, z dokładnością do milisekund lub mniejszą. Domyślne maksymalne przyszłe odchylenie zegara wynosi 30 sekund.
Monitoruj diagnostykę pod kątem opóźnionych lub zduplikowanych wejść oraz brakującego kontekstu. Aby zobaczyć limity próbki i reguły duplikowania, sprawdź plik contracts.json.
Przechowywaj w JSON buforowane znaczniki czasu i identyfikatory. Wyjściowe metadane MQTT pochodzą ze zdarzenia nadejścia, które udostępnia wynik, niekoniecznie z oryginalnego odczytu.

Dopasowywanie prognoz z opóźnionymi pomiarami według identyfikatora

Dopasuj predykcję 203 do późniejszego pomiaru 200 dla aktywa line-1, sztuki part-42. Residuum to pomiar minus przewidywanie: 200 - 203 = -3.

Schemat jednego źródła MQTT przekazującego prognozy i pomiary do niestandardowego modułu dopasowującego ID. Dopasowuje line-1 i part-42 oraz publikuje residual -3 i targetError 2.

Dopasuj według identyfikatorów aktywów i elementów. Dołącz identyfikator serii lub partii, jeśli ID elementów się powtarza. Zachowaj oczekujące rekordy dla dowolnej z dwóch kolejności nadejścia, ogranicz ich wiek i liczbę oraz zachowaj znaczniki czasowe i wersję modelu z prognozy.

Niestandardowa próbka służy do dopasowywania rekordów i obliczania residual oraz targetError. Residuum mierzy błąd predykcji; przy celu 198, oddzielny błąd docelowy to 200 - 198 = 2.

Wdróż przykład korelacji ID

Środowisko operacji obsługuje jedynie tworzenie i wyświetlanie artefaktów grafu z usługi ACR oraz, w przypadku wbudowanych transformacji, mcr.microsoft.com. Nie może wybrać ani wyświetlić artefaktu GHCR tej próbki przez factory-artifacts. Użyj zakładki Azure CLI lub Bicep, aby wdrożyć dokładnie ten przykład.

Dla osobno ponownie opublikowanego wariantu ACR z skonfigurowanym punktem końcowym rejestru ACR:

Utwórz graf przepływu danych o nazwie correlate-predictions przy użyciu istniejącego profilu default.
Skonfiguruj źródło za pomocą factory-mqtt oraz tematów factory/predictions i factory/measurements.
Wybierz Dodaj przekształcenie wykresu (opcjonalnie). W obszarze Wybór wykresu wybierz ponownie opublikowany graph-correlate-by-id:1.0.0-rc.4 artefakt i wybierz Zastosuj.
Skonfiguruj miejsce docelowe za pomocą factory-mqtt i tematufactory/matches, a następnie wybierz Zapisz. Środowisko Operations Experience automatycznie generuje połączenia węzłów artefaktu.

Zapisz te właściwości pełnego grafu jako 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"}}
  ]
}

Zastosuj do wykresu:

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

Zapisz ten pełny szablon jako main.bicep. Odnosi się do istniejącej instancji, niestandardowej lokalizacji i profilu:

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

Wdróż szablon:

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

Ważna

Korzystanie z manifestów wdrożenia platformy Kubernetes nie jest obsługiwane w środowiskach produkcyjnych i powinno być używane tylko do debugowania i testowania.

Azure Samples zapewnia kompletne correlate-predictions.yaml z własnymi tematami i opublikowanym odwołaniem do wykresu. Oddzielny align-by-time.yaml zawiera przykład ze znacznikiem czasu.

Sklonuj repozytorium próbek i wykonaj następujące polecenia po spełnieniu wspólnych wymagań wstępnych:

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

Manifesty te wykorzystują default istniejący profil i punkt końcowy factory-mqtt w azure-iot-operations. Manifest rejestru opisuje współdzielony factory-artifacts endpoint; jeśli jest już skonfigurowany, jego krok aplikacji jest idempotentny.

Korelacja ID wymaga, aby przewidywania zawierały kind: "prediction", assetId, pieceId, dodatnią liczbę całkowitą modelVersion oraz skończoną wartość liczbową predictedWeight. Pomiary wymagają kind: "measurement", tych samych identyfikatorów aktywów i elementów oraz skończonych liczb measuredWeight i targetWeight. Używaj oddzielnych pól predictedAt i measuredAt dla opcjonalnych znaczników czasu opisujących różne momenty. Wspólne pola niekluczowe muszą mieć równe wartości, inaczej dopasowanie się nie powiedzie.

Moduł niestandardowy zarządza oczekującymi meczami, limitami, obsługą duplikatów oraz wynikami ukończonych meczów. Wykres publikuje dopasowania do odrębnego tematu factory/matches.

Syntetyczne komunikaty wejściowe (NDJSON): publikuj pierwszą linię do factory/predictions, a drugą później do factory/measurements. Prognoza zawiera oryginalne wartości cech do ponownego przeszkolenia modelu.

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

Sama predykcja nie daje żadnego dopasowania. Gdy pomiar dotrze, spodziewaj się tego ładunku danych na factory/matches, w tym oryginalnej wersji modelu i wartości cech:

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

Obie kolejności przybycia się sprawdzają. Domyślnie wpisy oczekujące wygasają 60 sekund po nadejściu pierwszej strony; powtórki nie przedłużają tego czasu. Prawidłowe nadejścia lub {"kind":"tick","assetId":"line-1"} zaufane wiadomości wywołują czyszczenie; nie ma timera działającego w tle.

Ukończone mecze są domyślnie deduplikowane przez 60 sekund, a nie na zawsze. Ponowne uruchomienie powoduje utratę zarówno oczekujących, jak i ukończonych wpisów. Na potrzeby odzyskiwania należy deduplikować w zewnętrznym trwałym odbiorniku według zasobu, kawałka i opcjonalnego identyfikatora uruchomienia. Do ponownego trenowania zachowaj oryginalne wartości cech używane przez każdą predykcję.

Opinia

Czy ta strona była pomocna?

Ostatnia aktualizacja o 2026-10-29