10 Messwerte zu einem Ergebnis mitteln, um Schwankungen und das Nachrichtenvolumen zu reduzieren. Für Werte 10 ist der Mittelwert 9, 11, 10, 10, 9, 11, 10, 10, 9, 11.
Voraussetzungen
Erfüllen Sie die gemeinsamen Beispielvoraussetzungen, einschließlich des factory-mqtt Profils und des default Endpunkts. Verwenden Sie YAML zum Debuggen und Azure CLI oder Bicep für das Produktionsmanagement.
Verwenden Sie den vorhandenen integrierten Registrierungs-Endpunkt namens default. Für die Umgebungsvariablen setzen Sie AIO_INSTANCE_NAME und RESOURCE_GROUP auf die Namen Ihrer vorhandenen Instanz und Ressourcengruppe. Legen Sie für Bicep außerdem CUSTOM_LOCATION_NAME auf den Namen Ihres vorhandenen benutzerdefinierten Speicherorts fest.
Das integrierte Fenster erzeugt einen Durchschnitt pro nicht überlappendem Batch, nicht nach jedem Lesevorgang. Sie warten, bis der Batch geschlossen wird, und ein Timeout kann einen unvollständigen Batch schließen.
Für die Konfigurationssyntax siehe Count-based window und Boundary behavior. Die Einstellung messageInCurrent schließt die zehnte Nachricht in den Batch ein, den sie schließt.
Dieser Graph sendet nur die Messwerte eines Assets an das Fenster.
Der Filter verhindert, dass andere Assets in denselben Batch gemittelt werden.
Die zehn Werte sind separate Nachrichten, kein Array in einer Nachricht. Ein Asset ID-Feld teilt ein Fenster nicht automatisch auf; halte auch nicht zusammenhängende Signale und Einheiten getrennt.
Öffnen Sie in der Betriebsumgebung Ihre Instanz, und wählen Sie Datenflussgraph>Datenflussgraph erstellen aus.
Benennen Sie das Diagramm average-line-1 und wählen Sie das default Profil aus.
Fügen Sie eine Quelle mit dem Namen source mithilfe des Endpunkts factory-mqtt und des Themas factory/values hinzu.
Fügen Sie eine Filtertransformation namens only-line-1 hinzu, wobei Sie azureiotoperations/graph-dataflow-filter:1.1.0 aus dem Registry-Endpunkt default verwenden. Füge eine Regel mit Eingabe assetId und Ausdruck $1 != "line-1"hinzu. Ein Ergebnis „true“ verwirft die Nachricht.
Füge eine Fenstertransformation namens batch unter Verwendung von azureiotoperations/graph-dataflow-window:1.1.0 aus dem Registry-Endpunkt default hinzu. Setze die Nachrichtenanzahl auf 10 und das Verhalten der Grenznachrichten auf messageInCurrent. Lass den Timeout auf dem Standardwert von 60 Sekunden.
Fügen Sie diese Akkumulationsregeln dem Fenster hinzu:
| Input |
Ausgabe |
Ausdruck |
assetId |
assetId |
first($1) |
value |
meanValue |
average($1) |
value |
readingCount |
count($1) |
Fügen Sie ein Ziel mit dem Namen factory-mqtt unter Verwendung des Endpunkts factory/means und des Themas destination hinzu.
Verbinden Sie source → only-line-1 → batch → destination, und wählen Sie dann Speichern aus, um den aktivierten Graphen bereitzustellen.
Speichere diese vollständigen Grapheigenschaften als graph.json. Der Befehl verwendet das bestehende default Profil.
{
"mode": "Enabled",
"nodes": [
{
"name": "source",
"nodeType": "Source",
"sourceSettings": {
"endpointRef": "factory-mqtt",
"dataSources": ["factory/values"]
}
},
{
"name": "only-line-1",
"nodeType": "Graph",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-filter:1.1.0",
"configuration": [
{
"key": "rules",
"value": "{\"filter\":[{\"inputs\":[\"assetId\"],\"expression\":\"$1 != \\\"line-1\\\"\"}]}\n"
}
]
}
},
{
"name": "batch",
"nodeType": "Graph",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-window:1.1.0",
"configuration": [
{
"key": "count",
"value": "{\"type\":\"messageCount\",\"maxMessageCount\":10,\"boundaryMessage\":\"messageInCurrent\"}"
},
{
"key": "rules",
"value": "{\n \"accumulate\": [\n {\"inputs\":[\"assetId\"],\"output\":\"assetId\",\"expression\":\"first($1)\"},\n {\"inputs\":[\"value\"],\"output\":\"meanValue\",\"expression\":\"average($1)\"},\n {\"inputs\":[\"value\"],\"output\":\"readingCount\",\"expression\":\"count($1)\"}\n ]\n}\n"
}
]
}
},
{
"name": "destination",
"nodeType": "Destination",
"destinationSettings": {
"endpointRef": "factory-mqtt",
"dataDestination": "factory/means"
}
}
],
"nodeConnections": [
{ "from": { "name": "source" }, "to": { "name": "only-line-1" } },
{ "from": { "name": "only-line-1" }, "to": { "name": "batch" } },
{ "from": { "name": "batch" }, "to": { "name": "destination" } }
]
}
az iot ops dataflowgraph apply \
--name average-line-1 \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP \
--config-file graph.json
Speichere diese Vorlage als main.bicep. Es verweist auf die bestehende Instanz, den benutzerdefinierten Standort und das default-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: 'average-line-1'
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
mode: 'Enabled'
nodes: [
{
name: 'source'
nodeType: 'Source'
sourceSettings: {
endpointRef: 'factory-mqtt'
dataSources: [ 'factory/values' ]
}
}
{
name: 'only-line-1'
nodeType: 'Graph'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-filter:1.1.0'
configuration: [
{
key: 'rules'
value: '{"filter":[{"inputs":["assetId"],"expression":"$1 != \\"line-1\\""}]}\n'
}
]
}
}
{
name: 'batch'
nodeType: 'Graph'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-window:1.1.0'
configuration: [
{
key: 'count'
value: '{"type":"messageCount","maxMessageCount":10,"boundaryMessage":"messageInCurrent"}'
}
{
key: 'rules'
value: '{\n "accumulate": [\n {"inputs":["assetId"],"output":"assetId","expression":"first($1)"},\n {"inputs":["value"],"output":"meanValue","expression":"average($1)"},\n {"inputs":["value"],"output":"readingCount","expression":"count($1)"}\n ]\n}\n'
}
]
}
}
{
name: 'destination'
nodeType: 'Destination'
destinationSettings: {
endpointRef: 'factory-mqtt'
dataDestination: 'factory/means'
}
}
]
nodeConnections: [
{
from: { name: 'source' }
to: { name: 'only-line-1' }
}
{
from: { name: 'only-line-1' }
to: { name: 'batch' }
}
{
from: { name: 'batch' }
to: { name: 'destination' }
}
]
}
}
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.
Speichern Sie das folgende Manifest als average-line-1.yaml.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
name: average-line-1
namespace: azure-iot-operations
spec:
profileRef: default
mode: Enabled
nodes:
- name: source
nodeType: Source
sourceSettings:
endpointRef: factory-mqtt
dataSources: [factory/values]
- name: only-line-1
nodeType: Graph
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-filter:1.1.0
configuration:
- key: rules
value: |
{"filter":[{"inputs":["assetId"],"expression":"$1 != \"line-1\""}]}
- name: batch
nodeType: Graph
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-window:1.1.0
configuration:
- key: count
value: '{"type":"messageCount","maxMessageCount":10,"boundaryMessage":"messageInCurrent"}'
- key: rules
value: |
{
"accumulate": [
{"inputs":["assetId"],"output":"assetId","expression":"first($1)"},
{"inputs":["value"],"output":"meanValue","expression":"average($1)"},
{"inputs":["value"],"output":"readingCount","expression":"count($1)"}
]
}
- name: destination
nodeType: Destination
destinationSettings:
endpointRef: factory-mqtt
dataDestination: factory/means
nodeConnections:
- from: { name: source }
to: { name: only-line-1 }
- from: { name: only-line-1 }
to: { name: batch }
- from: { name: batch }
to: { name: destination }
kubectl apply -f average-line-1.yaml
Veröffentlichen Sie jede Zeile im folgenden Beispiel als separate Nachricht an factory/values:
{"assetId":"line-1","value":9}
{"assetId":"line-1","value":11}
{"assetId":"line-1","value":10}
{"assetId":"line-1","value":10}
{"assetId":"line-1","value":9}
{"assetId":"line-1","value":11}
{"assetId":"line-1","value":10}
{"assetId":"line-1","value":10}
{"assetId":"line-1","value":9}
{"assetId":"line-1","value":11}
Der Filter schließt andere Assets aus. Wenn es keinen früheren partiellen Batch gibt, schließt die zehnte Messung das Fenster und erzeugt diese Nutzlast auf factory/means (100 / 10 = 10):
{"assetId":"line-1","meanValue":10,"readingCount":10}
Senden Sie die zehn Messwerte innerhalb des Standard-60-Sekunden-Timeouts für das angegebene Zehn-Nachrichten-Ergebnis. Ein Timeout kann einen kleineren Batch veröffentlichen, was durch readingCount sichtbar gemacht wird. Überprüfen Sie, dass Sensor-Payloads numerische Werte enthalten, bevor Sie den Mittelwert verwenden.
Bei zehn Messwerten pro Sekunde dauert das Sammeln von zehn Messwerten etwa eine Sekunde; langsamere Eingaben erhöhen die Erfassungsverzögerung. Um das Standardtimeout zu ändern, konfigurieren Sie eine explizite dauerbasierte Schließungsbedingung, und berücksichtigen Sie unvollständige Batches nachgelagert.
Für einen geglätteten Wert nach jeder Messung siehe das Beispiel für den exponentiell gleitenden Durchschnitt in den Azure Samples Deployment-Anweisungen.
Verwandte Inhalte