Aggregieren von Daten im Laufe der Zeit in Datenflussdiagrammen

Von Bedeutung

Diese Seite enthält Anweisungen zum Verwalten von Azure IoT Operations Komponenten mithilfe von Kubernetes-Bereitstellungsmanifesten, die sich in PREVIEW befinden. Dieses Feature wird mit einigen Einschränkungen bereitgestellt und sollte nicht für Produktionsworkloads verwendet werden.

Die zusätzlichen Nutzungsbestimmungen für Microsoft Azure-Vorschauen enthalten rechtliche Bedingungen. Sie gelten für diejenigen Azure-Features, die sich in der Beta- oder Vorschauversion befinden oder aber anderweitig noch nicht zur allgemeinen Verfügbarkeit freigegeben sind.

Eine Fenstertransformation sammelt Nachrichten über ein festes Zeitintervall und erzeugt eine einzelne Ausgabenachricht mit aggregierten Werten. Anstatt jede einzelne Messung weiterzuleiten, können Sie Statistiken wie Mittelwerte, Mindestwerte oder Anzahlen berechnen und ein konsolidiertes Ergebnis an eine nachgelagerte Stelle senden.

Voraussetzungen

Eine Azure IoT Operations Instanz, die auf einem Arc-fähigen Kubernetes-Cluster bereitgestellt wird. Weitere Informationen finden Sie unter Deploy Azure IoT Operations.
Ein Standardregistrierungsendpunkt mit dem Namen default , auf den mcr.microsoft.com verweist, wird während der Bereitstellung automatisch erstellt.

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.

Funktionsweise der Fenstertransformation

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

Verzögerung: Legt die Dauer des Sturzfensters fest. Eingehende Nachrichten werden einer Fenstergrenze basierend auf ihrem Zeitstempel zugewiesen.
Akkumulation: Wendet die Aggregationsregeln an, wenn das Fenster geschlossen wird. Alle Nachrichten im Fenster werden auf eine einzelne Ausgabemeldung reduziert.

Hinweis

Der Verzögerungsschritt richtet Nachrichtenzeitstempel an Fensterbegrenzungen aus. Wenn eine Nachricht 7 Sekunden nach Beginn eines 10-Sekunden-Fensters eintrifft, wird sie der 10-Sekunden-Marke zugeordnet.

Konfigurieren der Fensterdauer

Der Verzögerungsschritt steuert, wie lange jedes Kippfenster andauert. Der Konfigurationsschlüssel lautet delay (nicht rules).

Legen Sie in der Fenstertransformationskonfiguration die Fensterdauer in Sekunden fest. Legen Sie ihn 30 z. B. für ein 30-Sekunden-Sturzfenster fest.

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

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

Eigentum	Typ	Beschreibung
`type`	Schnur	Muss `"duration"` sein.
`delaySeconds`	uint64	Die Größe des gleitenden Fensters in Sekunden. Muss größer als 0 sein.

Definieren von Akkumulationsregeln

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

Fügen Sie in der Fenstertransformationskonfiguration Akkumulationsregeln hinzu. Geben Sie für jede Regel Folgendes an:

Setting	Beschreibung
Eingabe	Das zu aggregierende Feld (z. B. `temperature`).
Output	Der Name des Ausgabefelds (z. B `avgTemperature`. ).
Ausdruck	Die Aggregationsfunktion (z. B. `average($1)`).

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

{
  "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, min und max Funktionen überspringen stillschweigend nicht-numerische Werte.

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

Kombinieren von Aggregationen

Sie können mehrere Aggregationsfunktionen in einem einzelnen Ausdruck kombinieren:

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

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

- 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

Eine vollständige Fensterkonfiguration, die Temperaturstatistiken über ein 30-Sekunden-Fenster berechnet.

Wenn das Fenster diese drei Nachrichten empfängt:

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

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. Legen Sie die Fensterdauer auf 30 Sekunden fest. Konfigurieren Sie Akkumulationsregeln für Mittelwert, Min., Max., Anzahl und Bereich des temperature Felds.
Fügen Sie ein Ziel hinzu, das an telemetry/aggregated sendet.

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2025-10-01' = {
  name: 'temperature-window'
  parent: dataflowProfile
  properties: {
    profileRef: dataflowProfileName
    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.0.0'
          configuration: [
            {
              key: 'delay'
              value: '{"type":"duration","delaySeconds":30}'
            }
            {
              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' } }
    ]
  }
}

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.0.0
        configuration:
          - key: delay
            value: |
              {
                "type": "duration",
                "delaySeconds": 30
              }
          - 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

Transformieren von Daten mit Karte

Feedback

War diese Seite hilfreich?

Last updated on 2026-03-25