Oblicz prognozę, stosując znany wzór do każdego odczytu. Na przykład weight = 2 * speed + 0.5 * temperature + 173 przewiduje wagę 203, gdy prędkość wynosi 10, a temperatura wynosi 20. Obliczenie wykonuje wbudowana transformacja mapy.
Przykład wykorzystuje zdefiniowaną formułę i wartości syntetyczne.
Powiązane niezależne wzorce obejmują dopasowanie i zatwierdzanie aktualizacji oraz dostarczanie zatwierdzonych współczynników za pomocą MQTT.
Prerequisites
Uzupełnij wspólne przykładowe wymagania wstępne, w tym default profil i factory-mqtt punkt końcowy. Używaj YAML do debugowania oraz Azure CLI lub Bicep do zarządzania środowiskiem produkcyjnym.
Do wbudowanych transformacji używaj istniejącego default endpointu rejestru. Dla wdrożeń CLI ustaw AIO_INSTANCE_NAME i RESOURCE_GROUP na swoją istniejącą instancję i grupę zasobów. Dla wdrożeń Bicep można również ustawić CUSTOM_LOCATION_NAME na istniejącą, niestandardową lokalizację instancji. Zakładki wdrożeń odnoszą się do tych współdzielonych zasobów; nie tworzą ich.
Wybierz skąd pochodzą współczynniki
| Źródło współczynnika |
Ścieżka przetwarzania |
Użyj, gdy |
| Dołączone do każdej wiadomości |
Odczyt ze współczynnikami → mapa → predykcja |
Producent już dostarcza model i jego wersję |
| Przechowywane oddzielnie według składnika |
Czytanie → wzbogacanie → mapowanie → przewidywanie |
Aplikacja utrzymuje zatwierdzone współczynniki niezależnie od komunikatów sensorów |
Wybierz jeden z dwóch pełnych wykresów. Wzbogacenie dostarcza przechowywane parametry; map ocenia formułę. Obie ścieżki zachowują identyfikację elementu, cechy oraz wersję modelu.
Oceń współczynniki dostarczone w wiadomości
Gdy jeden rekord zawiera dane wejściowe i współczynniki, transformacja mapowania ocenia speed * speedCoefficient + temperature * temperatureCoefficient + intercept.
Reguła symbolu wieloznacznego w poniższym przykładzie zachowuje oryginalne pola. Jeśli dane wejściowe przychodzą osobno, użyj wyszukiwania kontekstu lub korelacji , aby złożyć rekord.
Modele, które wymagają środowiska uruchomieniowego lub algorytmu wykraczającego poza język wyrażeń, wymagają niestandardowego WASM lub zewnętrznej usługi.
Wdroż ten wykres, stosując jedną z poniższych metod. Subskrybuj factory/predicted-values, a następnie publikuj następujące dane wejściowe do factory/features-with-model.
W doświadczeniu operacyjnym otwórz swoją instancję i wybierz Wykres przepływu danych>Utwórz wykres przepływu danych.
Nadaj wykresowi nazwę predict-with-coefficients i wybierz profil default
Dodaj źródło source z endpointem factory-mqtt i tematem factory/features-with-model.
Dodaj transformację mapy predict, za pomocą azureiotoperations/graph-dataflow-map:1.0.0 z punktu końcowego rejestru default. Konfiguruj te reguły po kolei:
| Dane wejściowe (w kolejności) |
Output |
Expression |
* |
* |
Pozostaw pole puste |
speed, temperature, , speedCoefficient, temperatureCoefficientintercept |
predictedWeight |
$1 * $3 + $2 * $4 + $5 |
Dodaj cel destination z endpointem factory-mqtt i tematem factory/predicted-values.
Połącz source → predict → destination i wybierz Zapisz.
Zapisz pełne właściwości grafu jako graph.json. Reguły transformacji to escapowane ciągi JSON w configuration.value.
{
"mode": "Enabled",
"nodes": [
{
"name": "source",
"nodeType": "Source",
"sourceSettings": {
"endpointRef": "factory-mqtt",
"dataSources": ["factory/features-with-model"]
}
},
{
"name": "predict",
"nodeType": "Graph",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"speed\",\"temperature\",\"speedCoefficient\",\"temperatureCoefficient\",\"intercept\"],\"output\":\"predictedWeight\",\"expression\":\"$1 * $3 + $2 * $4 + $5\"}]}"
}
]
}
},
{
"name": "destination",
"nodeType": "Destination",
"destinationSettings": {
"endpointRef": "factory-mqtt",
"dataDestination": "factory/predicted-values"
}
}
],
"nodeConnections": [
{"from": {"name": "source"}, "to": {"name": "predict"}},
{"from": {"name": "predict"}, "to": {"name": "destination"}}
]
}
Zastosuj graf do profilu istniejącej instancji default:
az iot ops dataflowgraph apply \
--name predict-with-coefficients \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP \
--config-file graph.json
Zapisz jako main.bicep. Wszystkie zasoby nadrzędne już istnieją w grupie zasobów.
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: 'predict-with-coefficients'
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
nodes: [
{
name: 'source'
nodeType: 'Source'
sourceSettings: {
endpointRef: 'factory-mqtt'
dataSources: ['factory/features-with-model']
}
}
{
name: 'predict'
nodeType: 'Graph'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["speed","temperature","speedCoefficient","temperatureCoefficient","intercept"],"output":"predictedWeight","expression":"$1 * $3 + $2 * $4 + $5"}]}'
}
]
}
}
{
name: 'destination'
nodeType: 'Destination'
destinationSettings: {
endpointRef: 'factory-mqtt'
dataDestination: 'factory/predicted-values'
}
}
]
nodeConnections: [
{ from: { name: 'source' }, to: { name: 'predict' } }
{ from: { name: 'predict' }, to: { name: 'destination' } }
]
}
}
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 jako predict-with-coefficients.yaml.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
name: predict-with-coefficients
namespace: azure-iot-operations
spec:
profileRef: default
mode: Enabled
nodes:
- name: source
nodeType: Source
sourceSettings:
endpointRef: factory-mqtt
dataSources: [factory/features-with-model]
- name: predict
nodeType: Graph
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{"inputs":["*"],"output":"*"},
{
"inputs":["speed","temperature","speedCoefficient","temperatureCoefficient","intercept"],
"output":"predictedWeight",
"expression":"$1 * $3 + $2 * $4 + $5"
}
]
}
- name: destination
nodeType: Destination
destinationSettings:
endpointRef: factory-mqtt
dataDestination: factory/predicted-values
nodeConnections:
- from: { name: source }
to: { name: predict }
- from: { name: predict }
to: { name: destination }
kubectl apply -f predict-with-coefficients.yaml
Dane wejściowe dotyczące factory/features-with-model:
{"assetId":"line-1","pieceId":"part-42","modelVersion":8,"speed":10,"temperature":20,"speedCoefficient":2,"temperatureCoefficient":0.5,"intercept":173}
Dane wyjściowe w factory/predicted-values zachowują pola wejściowe i dodają predictedWeight:
{"assetId":"line-1","pieceId":"part-42","modelVersion":8,"speed":10,"temperature":20,"speedCoefficient":2,"temperatureCoefficient":0.5,"intercept":173,"predictedWeight":203}
Odczytaj współczynniki z magazynu stanów
Użyj wzbogacenia, aby odczytać cały rekord active-model pod kluczem factory-active-models, a następnie ocenić formułę za pomocą transformacji mapy.
Dla samodzielnego dema użyj modułu zapisu state-store do utworzenia rekordu v8; nie jest wymagany żaden krok dopasowania. Zegary writera i brokera muszą być zsynchronizowane. Poczekaj na pasującą +OK odpowiedź przed publikacją odczytów.
Wdróż ten wykres, stosując jedną z poniższych metod. Subskrybuj factory/predictions, a następnie publikuj to wejście do factory/features:
{"assetId":"line-1","pieceId":"part-43","speed":10,"temperature":20}
W modelu v8 zainicjalizowanym ziarnem wyjście ma predictedWeight: 203 i modelVersion: 8.
Wykres wyświetla zatwierdzony rekord dla każdej prognozy.
W doświadczeniu operacyjnym otwórz swoją instancję i wybierz Wykres przepływu danych>Utwórz wykres przepływu danych.
Nadaj wykresowi nazwę predict-with-approved-model i wybierz profil default
Dodaj źródło source z endpointem factory-mqtt i tematem factory/features.
Dodaj transformację read-modelmapy, używając azureiotoperations/graph-dataflow-map:1.0.0 z punktu końcowego rejestru default. W konfiguracji zbioru danych ustaw klucz magazynu stanu factory-active-models as model, dopasuj wejścia $source.assetId, $context.assetId oraz dopasuj wyrażenie $1 == $2. Konfiguruj te reguły mapowania w kolejności, pozostawiając wyrażenia puste:
| Input |
Output |
* |
* |
$context(model).modelVersion |
modelVersion |
$context(model).speedCoefficient |
speedCoefficient |
$context(model).temperatureCoefficient |
temperatureCoefficient |
$context(model).intercept |
intercept |
Dodaj transformację predictmapy , używając tego samego artefaktu i endpointu rejestru. Konfiguruj te reguły w kolejności:
| Dane wejściowe (w kolejności) |
Output |
Expression |
* |
* |
Pozostaw pole puste |
assetId |
kind |
"prediction" |
speed, temperature, , speedCoefficient, temperatureCoefficientintercept |
predictedWeight |
$1 * $3 + $2 * $4 + $5 |
Dodaj miejsce docelowe destination z endpointem factory-mqtt i tematem factory/predictions.
Połącz source → read-model → predict → destination i wybierz Zapisz.
Zapisz pełne właściwości grafu jako graph.json.
{
"mode": "Enabled",
"nodes": [
{
"name": "source",
"nodeType": "Source",
"sourceSettings": {
"endpointRef": "factory-mqtt",
"dataSources": ["factory/features"]
}
},
{
"name": "read-model",
"nodeType": "Graph",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"datasets\":[{\"key\":\"factory-active-models as model\",\"inputs\":[\"$source.assetId\",\"$context.assetId\"],\"expression\":\"$1 == $2\"}],\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"$context(model).modelVersion\"],\"output\":\"modelVersion\"},{\"inputs\":[\"$context(model).speedCoefficient\"],\"output\":\"speedCoefficient\"},{\"inputs\":[\"$context(model).temperatureCoefficient\"],\"output\":\"temperatureCoefficient\"},{\"inputs\":[\"$context(model).intercept\"],\"output\":\"intercept\"}]}"
}
]
}
},
{
"name": "predict",
"nodeType": "Graph",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"assetId\"],\"output\":\"kind\",\"expression\":\"\\\"prediction\\\"\"},{\"inputs\":[\"speed\",\"temperature\",\"speedCoefficient\",\"temperatureCoefficient\",\"intercept\"],\"output\":\"predictedWeight\",\"expression\":\"$1 * $3 + $2 * $4 + $5\"}]}"
}
]
}
},
{
"name": "destination",
"nodeType": "Destination",
"destinationSettings": {
"endpointRef": "factory-mqtt",
"dataDestination": "factory/predictions"
}
}
],
"nodeConnections": [
{"from": {"name": "source"}, "to": {"name": "read-model"}},
{"from": {"name": "read-model"}, "to": {"name": "predict"}},
{"from": {"name": "predict"}, "to": {"name": "destination"}}
]
}
az iot ops dataflowgraph apply \
--name predict-with-approved-model \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP \
--config-file graph.json
Zapisz jako main.bicep.
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: 'predict-with-approved-model'
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
nodes: [
{
name: 'source'
nodeType: 'Source'
sourceSettings: {
endpointRef: 'factory-mqtt'
dataSources: ['factory/features']
}
}
{
name: 'read-model'
nodeType: 'Graph'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"datasets":[{"key":"factory-active-models as model","inputs":["$source.assetId","$context.assetId"],"expression":"$1 == $2"}],"map":[{"inputs":["*"],"output":"*"},{"inputs":["$context(model).modelVersion"],"output":"modelVersion"},{"inputs":["$context(model).speedCoefficient"],"output":"speedCoefficient"},{"inputs":["$context(model).temperatureCoefficient"],"output":"temperatureCoefficient"},{"inputs":["$context(model).intercept"],"output":"intercept"}]}'
}
]
}
}
{
name: 'predict'
nodeType: 'Graph'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["assetId"],"output":"kind","expression":"\\"prediction\\""},{"inputs":["speed","temperature","speedCoefficient","temperatureCoefficient","intercept"],"output":"predictedWeight","expression":"$1 * $3 + $2 * $4 + $5"}]}'
}
]
}
}
{
name: 'destination'
nodeType: 'Destination'
destinationSettings: {
endpointRef: 'factory-mqtt'
dataDestination: 'factory/predictions'
}
}
]
nodeConnections: [
{ from: { name: 'source' }, to: { name: 'read-model' } }
{ from: { name: 'read-model' }, to: { name: 'predict' } }
{ from: { name: 'predict' }, to: { name: 'destination' } }
]
}
}
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 jako predict-with-approved-model.yaml.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
name: predict-with-approved-model
namespace: azure-iot-operations
spec:
profileRef: default
mode: Enabled
nodes:
- name: source
nodeType: Source
sourceSettings:
endpointRef: factory-mqtt
dataSources: [factory/features]
- name: read-model
nodeType: Graph
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"datasets": [
{
"key":"factory-active-models as model",
"inputs":["$source.assetId","$context.assetId"],
"expression":"$1 == $2"
}
],
"map": [
{"inputs":["*"],"output":"*"},
{"inputs":["$context(model).modelVersion"],"output":"modelVersion"},
{"inputs":["$context(model).speedCoefficient"],"output":"speedCoefficient"},
{"inputs":["$context(model).temperatureCoefficient"],"output":"temperatureCoefficient"},
{"inputs":["$context(model).intercept"],"output":"intercept"}
]
}
- name: predict
nodeType: Graph
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{"inputs":["*"],"output":"*"},
{"inputs":["assetId"],"output":"kind","expression":"\"prediction\""},
{
"inputs":["speed","temperature","speedCoefficient","temperatureCoefficient","intercept"],
"output":"predictedWeight",
"expression":"$1 * $3 + $2 * $4 + $5"
}
]
}
- name: destination
nodeType: Destination
destinationSettings:
endpointRef: factory-mqtt
dataDestination: factory/predictions
nodeConnections:
- from: { name: source }
to: { name: read-model }
- from: { name: read-model }
to: { name: predict }
- from: { name: predict }
to: { name: destination }
kubectl apply -f predict-with-approved-model.yaml
Pierwsza mapa kopiuje wybrany model i wersję do rekordu.
Druga metoda oblicza na podstawie tych pól. Reguły mapowania odczytują wejście transformacji.
Aby również porównać prognozy przed i po aktualizacji, najpierw seeduj v7, zmieniając modelVersion writera na 7 oraz intercept na 175. Opublikuj odczyt, zapisz v8 i opublikuj odczyt dla nowego elementu.
| Przechowywany model |
Obliczenie dla prędkości 10, temperatury 20 |
Przewidywana masa |
| v7 |
10 * 2 + 20 * 0.5 + 175 |
205 |
| v8 |
10 * 2 + 20 * 0.5 + 173 |
203 |
Po obserwacji przez instancję v8, odczyt dla part-44 generuje takie wyjście na factory/predictions:
{"assetId":"line-1","pieceId":"part-44","speed":10,"temperature":20,"modelVersion":8,"speedCoefficient":2,"temperatureCoefficient":0.5,"intercept":173,"kind":"prediction","predictedWeight":203}
Pamięci podręczne wyszukiwania stanów mogą utrzymać aktywność v7 po pomyślnym zapisie. Sprawdź emitowane modelVersion, aby zobaczyć, kiedy instancja obserwuje v8. Przykład zakłada istnienie modelu dopasowania; dodaj ścieżkę odrzucenia lub odzyskiwania dla brakującego lub nieaktualnego kontekstu modelu.
Lookup korzysta z modelu dostępnego podczas przetwarzania. Dla modelu wybranego na podstawie znacznika czasu odczytu użyj dopasowania znaczników czasu.
Treści powiązane