Aggregierte Daten mit Fenstertransformationen in Datenflussdiagrammen

Eine Fenstertransformation gruppiert eingehende Nachrichten und erzeugt eine einzelne Ausgabenachricht mit aggregierten Werten, wenn das Fenster geschlossen wird. Anstatt jede Messung einzeln weiterzuleiten, können Sie Statistiken wie Durchschnitte, Mindestwerte oder Zählungen berechnen und ein konsolidiertes Ergebnis weiterleiten.

Derzeit kann ein Fenster je nach Dauer, Anzahl, Speicher oder Auslöserbedingungen geschlossen werden.

Eine Übersicht über Datenflussdiagramme und die Art und Weise, wie Transformationen in einer Pipeline erstellt werden, finden Sie in der Übersicht über Datenflussdiagramme.

Hinweis

Nicht auf Dauer basiertes Windowing erfordert azureiotoperations/graph-dataflow-window:1.1.0 oder höher.

Transformationen verwenden eine Ausdruckssprache, um Werte, Testbedingungen und Referenzfelder zu berechnen. Ausdrücke beziehen sich auf Eingaben nach Position, nicht nach Namen: Die erste Eingabe in der inputs Liste ist $1, die zweite ist $2, und so weiter. Integrierte Funktionen wie cToF konvertieren und bearbeiten diese Werte.

Für die vollständige Liste der Operatoren, Funktionen, Datentypen und Metadatenfelder siehe die Expressions-Referenz.

Fenstertransformationen fügen Aggregationsfunktionen wie average, min, und max, hinzu, die nur in Akkumulationsregeln verfügbar sind. Die vollständige Liste finden Sie unter Aggregationsfunktionen.

Voraussetzungen

Eine Instanz von Azure IoT Einsatz, die in einem Kubernetes-Cluster bereitgestellt wird. Weitere Informationen finden Sie unter Deploy Azure IoT Einsatz.

Ein Standardregistrierungsendpunkt mit dem Namen default , auf den mcr.microsoft.com verweist, wird während der Bereitstellung automatisch erstellt.

Die Azure CLI Beispiele in diesem Artikel verwenden Umgebungsvariablen, sodass Sie jeden Wert einmal festlegen und dann die Befehle as-iskopieren und einfügen können. Wenn Sie die Azure IoT Einsatz Codespaces-Umgebung aus dem Quickstart verwenden, sind diese Variablen bereits für Sie festgelegt und Sie können diesen Schritt überspringen. Ansonsten setze die folgenden Umgebungsvariablen in deiner Shell, bevor du die Befehle ausführst.

Die folgenden Skripte legen die am häufigsten verwendeten Umweltvariablen fest:

Umgebungsvariable	Beschreibung
`SUBSCRIPTION_ID`	Die ID des Abonnements, das Ihre Azure IoT Einsatz-Instanz enthält.
`RESOURCE_GROUP`	Der Name der Ressourcengruppe, die Ihre Azure IoT Einsatz-Instanz enthält.
`AIO_INSTANCE_NAME`	Der Name Ihrer Azure IoT Einsatz Instanz. Um deine Instanzen aufzulisten, führe `az iot ops list -o table`.
`CLUSTER_NAME`	Der Name des Azure Arc-fähigen Kubernetes-Clusters, der deine Instanz hostet.
`LOCATION`	Die Azure-Region zur Nutzung für neue Ressourcen, zum Beispiel `eastus`.

Bash
PowerShell

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Du musst nur die Variablen festlegen, die dieser Artikel verwendet. Dieser Artikel könnte zusätzliche Umgebungsvariablen für die von Ihnen gewählten Ressourcennamen verwenden. Der Artikel erklärt, wie man sie dort platziert, wo sie eingeführt werden.

Skalierungsbeschränkung für zustandsbehaftete Diagramme

Von Bedeutung

Fenster- und Gashebel-Transformationen sind zustandsvoll. Jede Instanz behält ihren eigenen Zustand, und Instanzen teilen diesen Zustand nicht miteinander. Wenn die Anzahl der Instanzen im Datenflussprofil größer als eins ist, verteilen geteilte Abonnements Nachrichten über die Instanzen, sodass jede Instanz nur eine Teilmenge der Nachrichten sieht. Eine Fenstertransformation berechnet dann Aggregationen wie Durchschnitte, Summen und Zählungen über einen partiellen Datensatz, und eine Throttle-Transformation setzt die konfigurierte Ratenbegrenzung unabhängig in jeder Instanz durch, anstatt über die gesamte Pipeline hinweg.

Setze die Anzahl der Datenflussprofil-Instanzen für jeden Datenflussgraphen, der eine Fenster- oder Throttle-Transformation verwendet, auf 1 . Zustandslose Datenflussgraphen, die ausschließlich Mapping-, Filter-, Verzweigungs- und Concatenate-Transformationen verwenden, können sicher höhere Instanzzahlen nutzen, um den Durchsatz zu erhöhen.

Wann eine Fenstertransformation verwendet werden soll

Verwenden Sie eine Fensterfunktion, wenn Sie Hochfrequenzsensordaten empfangen und das Volumen reduzieren möchten, bevor die Daten weitergeleitet werden. Zu den häufigen Szenarios gehören:

Berechnen von Durchschnittswerten: Ein Temperatursensor veröffentlicht jede Sekunde, aber Ihre Cloudanwendung benötigt nur einen 30-Sekunden-Durchschnitt.
Extremwerte verfolgen: Sie möchten die Mindest- und Maximaldruckwerte für jedes Intervall von einer Minute erreichen.
Zählen von Ereignissen: Sie müssen wissen, wie viele Türöffnungsereignisse in den letzten fünf Minuten aufgetreten sind.
Erstellen Sie Produktionschargen: Sie möchten Statistiken für jede feste Chargengröße berechnen, zum Beispiel für alle 100 Pakete, die von einer Fülllinie kommen.
Auf Zustandsänderungen reagieren: Sie möchten wissen, wann sich ein Betriebssignal ändert, zum Beispiel wenn ein Mischer von running zu wechselt draining.

Funktionsweise der Fenstertransformation

Die Fenstertransformation verfügt über zwei interne Schritte, die in einer Reihenfolge verbunden sind.

Fenster: Puffert Nachrichten, bis eine der konfigurierten Abschlussbedingungen eintritt.
Akkumulation: Wendet die Aggregationsregeln an, wenn das Fenster geschlossen wird. Die Transformation reduziert alle Nachrichten im Fenster auf eine einzige Ausgabenachricht.

Hinweis

Eine Fenstertransformation muss mindestens eine Schließungsbedingung konfigurieren: delay, count, , memoryoder triggers.

Konfigurieren Sie Fensterschließungsbedingungen

Ab der Version 1.1.0, fügt die Fenstertransformation neben dem bestehenden delay Schlüssel drei Konfigurationsschlüssel hinzu:

Konfigurationsschlüssel	Fenstertyp	Purpose
`delay`	Dauerbasiertes Zeitfenster	Schließen Sie das Fenster nach einer festen Dauer.
`count`	Zählbasiertes Fenster	Schließe das Fenster nach einer festen Anzahl von Nachrichten.
`memory`	Speicherbasiertes Fenster	Schließen Sie das Fenster, wenn die gepufferte Nutzlastgröße eine Grenze erreicht.
`triggers`	Triggerbasiertes Fenster	Schließen Sie das Fenster, wenn ein benutzerdefinierter Ausdruck als `true` ausgewertet wird.

Dauerbasiertes Zeitfenster

Nutze die Konfiguration, delay um das Fenster nach einer festen Dauer zu schließen. Diese Einstellung steuert, wie lange jedes Drehfenster dauert.

Hinweis

Der Verzögerungsschritt richtet die Zeitstempel der Nachrichten an den Fenstergrenzen aus. Wenn eine Nachricht 7 Sekunden nach Beginn eines 10-Sekunden-Fensters eintrifft, gehört sie zur 10-Sekunden-Grenze.

Hinweis

Wenn du nicht anbietest delay, nutzt das Fenster eine 60-Sekunden-Standard-Timeout als Sicherheitsventil.

Legen Sie in der Fenstertransformationskonfiguration die Fensterdauer in Sekunden fest. Stellen Sie es beispielsweise auf 30 ein, um ein 30-sekündiges Zeitfenster zu erhalten.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an. Füge diese Konfiguration den Fenstertransformationsknoten configuration in deiner graph.json Datei hinzu. Dann wenden Sie es mit az iot ops dataflowgraph apply an.

Die Regeln sind ein JSON-Objekt:

{
  "type": "duration",
  "delaySeconds": 30
}

Diese Regeln werden als Escapezeichenfolge im value-Feld abgelegt:

"configuration": [
  {
    "key": "delay",
    "value": "{\"type\":\"duration\",\"delaySeconds\":30}"
  }
]

Tip

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

configuration: [
  {
    key: 'delay'
    value: '{"type":"duration","delaySeconds":30}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "type": "duration",
  "delaySeconds": 30
}

Eigentum	Typ	Beschreibung
`type`	Schnur	Muss `"duration"`lauten.
`delaySeconds`	uint64	Die Anzahl der Sekunden, bevor das Fenster sich schließt. Muss größer als 0 sein.

Zählbasiertes Fenster

Verwenden Sie die Konfiguration, count um das Fenster nach einer festen Anzahl von Nachrichten zu schließen.

Setzen Sie in der Konfiguration der Fenstertransformation die Anzahl der Nachrichten auf 5 und stellen Sie das Verhalten bei Grenznachrichten auf messageInCurrent.

Die CLI wendet den gesamten Graphen aus einer Konfigurationsdatei an. Fügen Sie diese Konfiguration also der configuration des Window-Transformationsknotens in Ihrer graph.json-Datei hinzu. Dann wenden Sie es mit az iot ops dataflowgraph apply an.

Die Regeln sind ein JSON-Objekt:

{
  "type": "messageCount",
  "maxMessageCount": 5,
  "boundaryMessage": "messageInCurrent"
}

Fügen Sie diese Regeln im value-Feld als Escapezeichenfolge hinzu:

"configuration": [
  {
    "key": "count",
    "value": "{\"type\":\"messageCount\",\"maxMessageCount\":5,\"boundaryMessage\":\"messageInCurrent\"}"
  }
]

Tip

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

configuration: [
  {
    key: 'count'
    value: '{"type":"messageCount","maxMessageCount":5,"boundaryMessage":"messageInCurrent"}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "type": "messageCount",
  "maxMessageCount": 5,
  "boundaryMessage": "messageInCurrent"
}

Eigentum	Typ	Beschreibung
`type`	Schnur	Muss `"messageCount"`lauten.
`maxMessageCount`	uint64	Anzahl der Nachrichten, die vor dem Schließen des Fensters gepuffert werden müssen. Muss größer als 0 sein.
`boundaryMessage`	Schnur	Ob die Nachricht, die das Fenster schließt, im aktuellen Fenster bleibt (`messageInCurrent`) oder das nächste Fenster startet (`messageInNext`).

Speicherbasiertes Fenster

Verwenden Sie die Konfiguration, memory um das Fenster zu schließen, wenn die gepufferte Nutzlastgröße ein Limit erreicht.

Stellen Sie in der Konfiguration der Fenstertransformation die Puffergröße auf 1048576 Byte ein und setzen Sie das Verhalten für Begrenzungsnachrichten auf messageInNext.

Die Regeln sind ein JSON-Objekt:

{
  "type": "bufferSize",
  "maxBufferBytes": 1048576,
  "boundaryMessage": "messageInNext"
}

Fügen Sie diese Regeln im value-Feld als Escapezeichenfolge hinzu:

"configuration": [
  {
    "key": "memory",
    "value": "{\"type\":\"bufferSize\",\"maxBufferBytes\":1048576,\"boundaryMessage\":\"messageInNext\"}"
  }
]

Tip

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

configuration: [
  {
    key: 'memory'
    value: '{"type":"bufferSize","maxBufferBytes":1048576,"boundaryMessage":"messageInNext"}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "type": "bufferSize",
  "maxBufferBytes": 1048576,
  "boundaryMessage": "messageInNext"
}

Eigentum	Typ	Beschreibung
`type`	Schnur	Muss `"bufferSize"`lauten.
`maxBufferBytes`	uint64	Maximale Gesamtzahl an Nutzlast-Bytes, bevor sich das Fenster schließt. Muss größer als 0 sein.
`boundaryMessage`	Schnur	Ob die Nachricht, die das Fenster schließt, im aktuellen Fenster bleibt (`messageInCurrent`) oder das nächste Fenster startet (`messageInNext`).

Triggerbasiertes Fenster

Verwenden Sie die triggers Konfiguration, wenn das Fenster basierend auf dem Nachrichteninhalt oder dem Ausführungszustand im aktuellen Fenster geschlossen werden soll.

In der Fenstertransformationskonfiguration fügen Sie eine Triggerregel mit Eingabefeld temperature, Ausdruck running_sum($1) + $1 > 100und Verhalten der Grenznachricht messageInCurrent hinzu.

Die Regeln sind ein JSON-Objekt:

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

Fügen Sie diese Regeln im value-Feld als Escapezeichenfolge hinzu:

"configuration": [
  {
    "key": "triggers",
    "value": "{\"type\":\"expression\",\"rules\":[{\"inputs\":[\"temperature\"],\"trigger\":\"running_sum($1) + $1 > 100\",\"boundaryMessage\":\"messageInCurrent\"}]}"
  }
]

Tip

Um die entkommene Zeichenkette zu erzeugen, speichere die Regeln in einer Datei wie rules.json, führe jq -c . rules.json, und füge die einzeilige Ausgabe ins Feld value ein.

configuration: [
  {
    key: 'triggers'
    value: '{"type":"expression","rules":[{"inputs":["temperature"],"trigger":"running_sum($1) + $1 > 100","boundaryMessage":"messageInCurrent"}]}'
  }
]

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

Eigentum	Erforderlich	Beschreibung
`type`	Ja	Muss `"expression"`lauten.
`rules`	Ja	Eine Vielzahl von Auslöserregeln. Regeln werden sequentiell pro Nachricht ausgewertet; Die erste Abgleichsregel schließt das Fenster.
`datasets`	No	Optionale Datensätze des Statusspeichers, die auf den Statusspeicher verweisen.

Jede Auslöserregel unterstützt diese Felder:

Eigentum	Erforderlich	Beschreibung
`inputs`	Ja	Array von Referenzen auf Eingabefelder. Der Ausdruck bindet an `$1`, `$2`, und so weiter.
`trigger`	Ja	Boolescher Ausdruck, der das Fenster schließt, wenn er zu `true` ausgewertet wird.
`boundaryMessage`	Ja	Ob die Nachricht, die das Fenster schließt, im aktuellen Fenster bleibt (`messageInCurrent`) oder das nächste Fenster startet (`messageInNext`).

Das Feld inputs unterstützt die gleiche Eingabesyntax, die auch anderswo in Datenflussgraphen verwendet wird, einschließlich einfacher Felder, ?? Standardwerte, ? $last, $context(key).field, und $metadata.*. Weitere Details zur Verwendung von $context(key) finden Sie unter Mit externen Daten anreichern.

Triggerausdrücke können die regulären Diagrammausdrucksfunktionen und die folgenden Ausführungsstatusfunktionen verwenden, die zurückgesetzt werden, wenn das Fenster geschlossen wird:

Funktion	Beschreibung
`running_sum($1)`	Kumulierte Summe von `$1` über vorherige Nachrichten im aktuellen Fenster.
`running_avg($1)`	Kumulativer Mittelwert von `$1` über vorherige Nachrichten.
`running_min($1)`	Minimalwert von `$1`, der in vorherigen Nachrichten angezeigt wurde. Gibt `$1` für die erste Nachricht zurück (das Minimum eines einzelnen Elements ist das Element selbst).
`running_max($1)`	Höchstwert von `$1` aus vorherigen Nachrichten. Gibt `$1` in der ersten Nachricht zurück (maximal ein Element).
`running_count($1)`	Anzahl der Nachrichten, in denen `$1` vorhanden war.
`running_count()`	Gesamtanzahl der Nachrichten (kein Feldfilter).
`first($1)`	Der erste nicht-leere Wert von `$1` im aktuellen Fenster. Gibt bei der ersten Nachricht `$1` zurück.
`changed($1)`	`true` wenn `$1` sich von seinem Wert in der vorherigen Nachricht unterscheidet. `false` bei der ersten Nachricht eines Fensters (kein vorheriger Wert zum Vergleich).
`prev($1)`	Der zuletzt verwendete nicht leere Wert von `$1` aus einer vorherigen Nachricht im aktuellen Fenster. Nachrichten, bei denen `$1` Leer war, werden übersprungen (der gespeicherte Wert wird nicht überschrieben). Gibt bei der ersten Nachricht eines Fensters `$1` zurück.

Hinweis

running_sum($1) und ähnliche Funktionen geben Werte aus zuvor verarbeiteten Nachrichten zurück. Verwenden Sie für die aktuelle Nachricht $1.

Beispiele für Auslöserregeln

Verwenden Sie diese Beispiele, um gemeinsame inputs Muster und trigger Muster in einem vollständigen Trigger-Konfigurationsobjekt zu erkennen:

Dieses Beispiel zeigt einen regulären Auslöser-Ausdruck. Das Fenster wird geschlossen, wenn der aktuelle temperature größer als 80 ist.

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "$1 > 80",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

Dieses Beispiel zeigt einen Trigger-Ausdruck, der running_sum($1) + $1 verwendet, um frühere Nachrichten im aktuellen Fenster mit der aktuellen Nachricht zu kombinieren und dann das Fenster zu schließen, sobald der Schwellenwert überschritten wird.

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature"],
      "trigger": "running_sum($1) + $1 > 100",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

Dieses Beispiel zeigt die nullsichere Eingabeverarbeitung mit temperature ?? 0, plus messageInNext , um die Grenznachricht im nächsten Fenster zu platzieren.

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["temperature ?? 0"],
      "trigger": "running_avg($1) > 80",
      "boundaryMessage": "messageInNext"
    }
  ]
}

Dieses Beispiel zeigt einen metadatenbasierten Trigger, bei dem das Fenster für einen bestimmten Themenwert aus $metadata.topicschließt.

{
  "type": "expression",
  "rules": [
    {
      "inputs": ["$metadata.topic"],
      "trigger": "$1 == \"telemetry/high-priority\"",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

Dieses Beispiel zeigt Trigger-Regeln mit Datensatzanreicherung: Es gleicht die Nachricht factoryId mit einer Zustandsspeicherzeile ab, liest shiftId aus $context(factory).shiftIdund schließt das Fenster, wenn sich dieser Verschiebungswert ändert (changed($1)).

{
  "type": "expression",
  "datasets": [
    {
      "key": "factory",
      "inputs": ["$source.factoryId", "$context.factoryId"],
      "expression": "$1 == $2"
    }
  ],
  "rules": [
    {
      "inputs": ["$context(factory).shiftId"],
      "trigger": "changed($1)",
      "boundaryMessage": "messageInCurrent"
    }
  ]
}

In diesem Beispiel wird erwartet, dass der durch factory dargestellte State-Store-Datensatz Felder wie factoryId und shiftId enthält.

Randverhalten

Die Einstellung boundaryMessage steuert, was mit der Nachricht geschieht, die dazu führte, dass ein Zähl-, Speicher- oder Trigger-basiertes Fenster geschlossen wurde:

messageInCurrent: Fügen Sie die Grenzmeldung im Schließen des Fensters ein.
messageInNext: Schließen Sie zuerst das aktuelle Fenster, dann starten Sie das nächste Fenster mit der Grenzmeldung.

Wenn messageInNext bei der ersten Nachricht in einem neuen Fenster ausgelöst wird, wird das Schließen des Fensters unterdrückt, damit kein leeres Fenster ausgegeben wird.

Hinweis

Das zeitbasierte Fenster verwendet boundaryMessage nicht. Dauergrenzen sind zeitbasiert, nicht nachrichtenbasiert, daher gibt es keine Grenzmeldung, die im aktuellen oder nächsten Fenster platziert werden muss.

Abschlussbedingungen kombinieren

Man kann Dauer, Anzahl, Speicher und Auslöserbedingungen im selben Diagramm kombinieren.

Die Dauer wird zeitlich gesteuert und vom Timer bewertet.
Für jede eingehende Nachricht werden nachrichtengetriebene Bedingungen in folgender Reihenfolge ausgewertet: Memory > Count > Trigger.
Innerhalb von triggers.rules werden Regeln nacheinander ausgewertet, und die erste zutreffende Regel angewendet.

Die Auswertungsreihenfolge in nachrichtengetriebenen Bedingungen ist für die Akkumulationsergebnisse wichtig, wenn eine Nachricht mehrere Bedingungen gleichzeitig erfüllt. Zum Beispiel, wenn memory verwendet messageInCurrent und count verwendet messageInNext, folgt eine Nachricht, die beide Bedingungen erfüllt, der Arbeitsspeicherkonfiguration. Die Nachricht bleibt im aktuellen Fenster und trägt zur Akkumulationsausgabe dieses Fensters bei.

Definieren von Akkumulationsregeln

Jede Akkumulationsregel gibt an, wie ein Fenster von Nachrichten in einen einzelnen Ausgabewert reduziert wird. Der Konfigurationsschlüssel lautet rules.

In der Fenstertransformationskonfiguration fügen Sie eine Akkumulationsregel mit Eingabe temperature, Ausgabe avgTemperatureund Aggregationsfunktion average($1)hinzu.

Die CLI wendet den gesamten Graphen aus einer einzigen Konfigurationsdatei an, fügen Sie dies also dem Knoten für die Fenstertransformation configuration in Ihrer graph.json hinzu und wenden Sie es mit az iot ops dataflowgraph apply an.

Die Regeln sind ein JSON-Objekt:

{
  "accumulate": [
    {
      "inputs": ["temperature"],
      "output": "avgTemperature",
      "expression": "average($1)"
    }
  ]
}

Diese Regeln werden als Escapezeichenfolge im value-Feld abgelegt:

{
  "key": "rules",
  "value": "{\"accumulate\":[{\"inputs\":[\"temperature\"],\"output\":\"avgTemperature\",\"expression\":\"average($1)\"}]}"
}

{
  key: 'rules'
  value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"}]}'
}

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

{
  "accumulate": [
    {
      "inputs": ["temperature"],
      "output": "avgTemperature",
      "expression": "average($1)"
    }
  ]
}

Eigentum	Erforderlich	Beschreibung
`inputs`	Ja	Liste der Feldpfade, die aus jeder eingehenden Nachricht gelesen werden sollen.
`output`	Ja	Feldpfad für das aggregierte Ergebnis. Jede Regel muss über eine eindeutige Ausgabe verfügen.
`expression`	Ja	Formel, die Eingabewerte über das Fenster auf einen einzelnen Skalar reduziert. Muss mindestens eine Aggregationsfunktion enthalten.
`description`	No	Menschenlesbare Beschreibung.

Im Gegensatz zu Kartenregeln ist expression für jede Akkumulationsregel erforderlich. Die Verwendung $1 allein ist ungültig, da sie auf eine Auflistung von Werten verweist, nicht auf einen einzelnen Skalar. Sie müssen sie in eine Aggregationsfunktion wie average($1) einschließen.

Aggregationsfunktionen

Funktion	Rückkehr	Verhalten des leeren Fensters
`average`	Mittelwert numerischer Werte	Fehler
`sum`	Summe numerischer Werte	0,0
`min`	Minimaler numerischer Wert	Fehler
`max`	Maximaler numerischer Wert	Fehler
`count`	Anzahl der Nachrichten, in denen das Feld vorhanden ist	0
`first`	Erster Wert im Fenster	Fehler
`last`	Letzter Wert im Fenster	Fehler

Jede Funktion verwendet eine einzelne Positionsvariable als Argument ($1 für die erste Eingabe, $2 für die zweite usw.).

Nicht-numerische Werte: Die , average, sum und min Funktionen überspringen stillschweigend nicht-numerische Werte.

Anwesenheitsbasierte Funktionen: count, first und last arbeiten mit der Feldpräsenz unabhängig vom Werttyp.

Kombinieren von Aggregationen

Kombinieren Sie mehrere Aggregationsfunktionen in einem einzigen Ausdruck:

Fügen Sie eine Regel mit den Eingaben temperature und humidity sowie dem Ausdruck average($1) + max($2) hinzu.

Die CLI wendet das gesamte Diagramm aus einer Konfigurationsdatei an, fügen Sie dies also an die entsprechende Stelle in Ihrem graph.json hinzu, und wenden Sie es mit az iot ops dataflowgraph apply:

{
  "inputs": [
    "temperature",
    "humidity"
  ],
  "output": "tempHumidityIndex",
  "expression": "average($1) + max($2)"
}

{
  inputs: [ 'temperature', 'humidity' ]
  output: 'tempHumidityIndex'
  expression: 'average($1) + max($2)'
}

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

- inputs:
    - temperature     # $1
    - humidity        # $2
  output: tempHumidityIndex
  expression: "average($1) + max($2)"

Um einen aggregierten Wert zu konvertieren, wenden Sie die Konvertierungsfunktion außerhalb der Aggregation an. Wandelt beispielsweise cToF(average($1)) die Durchschnittliche Temperatur in Fahrenheit um.

Jede Aggregationsfunktion muss direkt auf eine einzelne Positionsvariable verweisen. average($1) + max($2) ist gültig, aber average($1 + $2) nicht.

Unterschiede zwischen Kartenregeln

Fähigkeit	Kartierungsregeln	Akkumulationsregeln
Ausdruck erforderlich	No	Ja
Wildcardeingaben	Unterstützt	Nicht unterstützt
`$metadata` Zugriff	Unterstützt	Nicht unterstützt
`$context` Bereicherung	Unterstützt	Nicht unterstützt
`? $last`-Anweisung	Unterstützt	Nicht unterstützt
Ausgabeinhaltstyp	Entspricht Eingaben	Immer `application/json`

Vollständiges Konfigurationsbeispiel

Dieses Beispiel zeigt eine vollständige Fensterkonfiguration, die das Fenster nach 30 Sekunden, 5 Nachrichten, 1.048.576 gepufferten Bytes schließt, oder wenn running_sum($1) + $1 > 100. Das Beispiel setzt den boundaryMessage Wert auf messageInCurrent für die letzten drei Bedingungen, und das Fenster berechnet beim Schließen Temperaturstatistiken.

Welche Bedingung das Fenster schließt, hängt von der Nachrichtenzeit, der Anzahl, der Nutzlastgröße und dem Inhalt ab. Die folgenden Beispiele zeigen die resultierende Ausgabe für jede Schließbedingung.

Dauer endet

Wenn keine andere Bedingung zuerst ausgelöst wird und das Fenster nach Empfang dieser drei Nachrichten 30 Sekunden erreicht:

{ "temperature": 21.5 }
{ "temperature": 23.0 }
{ "temperature": 19.8 }

Die Ausgabemeldung lautet:

{
  "avgTemperature": 21.433333333333334,
  "minTemperature": 19.8,
  "maxTemperature": 23.0,
  "readingCount": 3,
  "tempRange": 3.2
}

Zählung endet

Wenn das Fenster diese fünf Nachrichten empfängt, bevor irgendeine andere Bedingung eintritt:

{ "temperature": 20.0 }
{ "temperature": 22.0 }
{ "temperature": 21.0 }
{ "temperature": 24.0 }
{ "temperature": 23.0 }

Die Ausgabemeldung lautet:

{
  "avgTemperature": 22.0,
  "minTemperature": 20.0,
  "maxTemperature": 24.0,
  "readingCount": 5,
  "tempRange": 4.0
}

Speicher wird geschlossen

Wenn die gepufferte Nutzlastgröße 1.048.576 Byte erreicht, bevor eine andere Bedingung eintritt, beispielsweise nach diesen beiden großen Nachrichten:

{ "temperature": 21.0, "payloadPad": "<large string>" }
{ "temperature": 22.5, "payloadPad": "<large string>" }

Die Ausgabemeldung lautet:

{
  "avgTemperature": 21.75,
  "minTemperature": 21.0,
  "maxTemperature": 22.5,
  "readingCount": 2,
  "tempRange": 1.5
}

Auslöser wird geschlossen

Wenn der Trigger-Ausdruck running_sum($1) + $1 > 100 vor allen anderen Bedingungen ausgelöst wird, zum Beispiel nach diesen drei Nachrichten:

{ "temperature": 40.0 }
{ "temperature": 35.0 }
{ "temperature": 30.0 }

Die Ausgabemeldung lautet:

{
  "avgTemperature": 35.0,
  "minTemperature": 30.0,
  "maxTemperature": 40.0,
  "readingCount": 3,
  "tempRange": 10.0
}

Erstellen Sie in der Betriebsumgebung ein Datenflussdiagramm mit einer Fenstertransformation:

Fügen Sie eine Quelle hinzu, die aus telemetry/temperature liest.
Fügen Sie eine Fenstertransformation hinzu. Konfigurieren Sie ein 30-sekündiges Dauerfenster, ein Limit von 5 Nachrichten, ein Puffergrößenlimit von 1.048.576 Byte und eine Triggerregel auf temperature mit Ausdruck running_sum($1) + $1 > 100. Legen Sie für die Bedingungen für Anzahl, Speicher und Trigger das Verhalten der Begrenzungsnachricht auf messageInCurrent fest. Fügen Sie Aggregationsregeln für Durchschnitt, Minimum, Maximum, Anzahl und Spannweite für das Feld temperature hinzu.
Fügen Sie ein Ziel hinzu, das an telemetry/aggregated sendet.

Die Azure CLI wendet ein Datenflussdiagramm aus einer einzelnen JSON-Konfigurationsdatei an. Erstellen Sie eine graph.json Datei mit den Diagrammeigenschaften. Speichere in der graph.json Datei die Regeln jeder Transformation im Feld value als entkommene JSON-Zeichenkette. Die lesbare Form der Regeln der einzelnen Transformationen finden Sie in der Vorgehensweise für diesen Transformationstyp.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "aggregate",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-window:1.1.0",
        "configuration": [
          {
            "key": "delay",
            "value": "{\"type\":\"duration\",\"delaySeconds\":30}"
          },
          {
            "key": "count",
            "value": "{\"type\":\"messageCount\",\"maxMessageCount\":5,\"boundaryMessage\":\"messageInCurrent\"}"
          },
          {
            "key": "memory",
            "value": "{\"type\":\"bufferSize\",\"maxBufferBytes\":1048576,\"boundaryMessage\":\"messageInCurrent\"}"
          },
          {
            "key": "triggers",
            "value": "{\"type\":\"expression\",\"rules\":[{\"inputs\":[\"temperature\"],\"trigger\":\"running_sum($1) + $1 > 100\",\"boundaryMessage\":\"messageInCurrent\"}]}"
          },
          {
            "key": "rules",
            "value": "{\"accumulate\":[{\"inputs\":[\"temperature\"],\"output\":\"avgTemperature\",\"expression\":\"average($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"minTemperature\",\"expression\":\"min($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"maxTemperature\",\"expression\":\"max($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"readingCount\",\"expression\":\"count($1)\"},{\"inputs\":[\"temperature\"],\"output\":\"tempRange\",\"expression\":\"max($1) - min($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/aggregated"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "aggregate"
      }
    },
    {
      "from": {
        "name": "aggregate"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Wenden Sie die Konfigurationsdatei an.

az iot ops dataflowgraph apply \
  --name temperature-window \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  name: 'temperature-window'
  parent: dataflowProfile
  properties: {
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [ 'telemetry/temperature' ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'aggregate'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-window:1.1.0'
          configuration: [
            {
              key: 'delay'
              value: '{"type":"duration","delaySeconds":30}'
            }
            {
              key: 'count'
              value: '{"type":"messageCount","maxMessageCount":5,"boundaryMessage":"messageInCurrent"}'
            }
            {
              key: 'memory'
              value: '{"type":"bufferSize","maxBufferBytes":1048576,"boundaryMessage":"messageInCurrent"}'
            }
            {
              key: 'triggers'
              value: '{"type":"expression","rules":[{"inputs":["temperature"],"trigger":"running_sum($1) + $1 > 100","boundaryMessage":"messageInCurrent"}]}'
            }
            {
              key: 'rules'
              value: '{"accumulate":[{"inputs":["temperature"],"output":"avgTemperature","expression":"average($1)"},{"inputs":["temperature"],"output":"minTemperature","expression":"min($1)"},{"inputs":["temperature"],"output":"maxTemperature","expression":"max($1)"},{"inputs":["temperature"],"output":"readingCount","expression":"count($1)"},{"inputs":["temperature"],"output":"tempRange","expression":"max($1) - min($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/aggregated'
        }
      }
    ]
    nodeConnections: [
      { from: { name: 'sensors' }, to: { name: 'aggregate' } }
      { from: { name: 'aggregate' }, to: { name: 'output' } }
    ]
  }
}

Von Bedeutung

Die Verwendung von Kubernetes-Bereitstellungsmanifesten wird in Produktionsumgebungen nicht unterstützt und sollte nur zum Debuggen und Testen verwendet werden.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: temperature-window
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/temperature

    - nodeType: Graph
      name: aggregate
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-window:1.1.0
        configuration:
          - key: delay
            value: |
              {
                "type": "duration",
                "delaySeconds": 30
              }
          - key: count
            value: |
              {
                "type": "messageCount",
                "maxMessageCount": 5,
                "boundaryMessage": "messageInCurrent"
              }
          - key: memory
            value: |
              {
                "type": "bufferSize",
                "maxBufferBytes": 1048576,
                "boundaryMessage": "messageInCurrent"
              }
          - key: triggers
            value: |
              {
                "type": "expression",
                "rules": [
                  {
                    "inputs": ["temperature"],
                    "trigger": "running_sum($1) + $1 > 100",
                    "boundaryMessage": "messageInCurrent"
                  }
                ]
              }
          - key: rules
            value: |
              {
                "accumulate": [
                  {
                    "inputs": ["temperature"],
                    "output": "avgTemperature",
                    "expression": "average($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "minTemperature",
                    "expression": "min($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "maxTemperature",
                    "expression": "max($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "readingCount",
                    "expression": "count($1)"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "tempRange",
                    "expression": "max($1) - min($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/aggregated

  nodeConnections:
    - from: { name: sensors }
      to: { name: aggregate }
    - from: { name: aggregate }
      to: { name: output }

Nächste Schritte

Feedback

War diese Seite hilfreich?

Last updated on 2026-08-14

Aggregierte Daten mit Fenstertransformationen in Datenflussdiagrammen

Voraussetzungen

Skalierungsbeschränkung für zustandsbehaftete Diagramme

Wann eine Fenstertransformation verwendet werden soll

Funktionsweise der Fenstertransformation

Konfigurieren Sie Fensterschließungsbedingungen

Dauerbasiertes Zeitfenster

Zählbasiertes Fenster

Speicherbasiertes Fenster

Triggerbasiertes Fenster

Beispiele für Auslöserregeln

Randverhalten

Abschlussbedingungen kombinieren

Definieren von Akkumulationsregeln

Aggregationsfunktionen

Kombinieren von Aggregationen

Unterschiede zwischen Kartenregeln

Vollständiges Konfigurationsbeispiel

Dauer endet

Zählung endet

Speicher wird geschlossen

Auslöser wird geschlossen

Nächste Schritte

Feedback

Zusätzliche Ressourcen