Zum Hauptinhalt springen

AggregateStreamData@1

Der Node AggregateStreamData@1 verdichtet die Spalten eines Stream-Data-Archivs zu Kennzahlen über einen Zeitraum — die Summe der Energie eines Monats, die maximale Datenqualität in diesem Monat. Er ist das Gegenstück zu GetStreamData@1, der die Zeilen selbst liefert.

Der Node schreibt ein Query-Ergebnis nach targetPath: eine Spalte je Kennzahl und, wenn groupBy verwendet wird, eine Zeile je Gruppe.

Adapter-Voraussetzungen​

Node-Konfiguration​

Zu den Feldern targetPath, targetValueWriteMode und targetValueKind siehe Überblick. Das Feld path wird in diesem Node nicht verwendet.

transformations:
  - type: AggregateStreamData@1
    archiveRtId: 68a1f0c5de73e7b175575401 # Runtime-ID des Archivs (muss aktiviert sein)
    aggregations: # Die zu bildenden Kennzahlen, mindestens eine
      - attributePath: Energy
        function: Sum
      - attributePath: DataQuality
        function: Maximum
    groupBy: # Spalten zur Gruppierung, leer ergibt eine einzige Ergebniszeile
      - rtId
    wellKnownNames: # Einschränkung auf Quell-Entitäten mit diesen Well-Known Names
      - METER-4711
    wellKnownNamesPath: $.meters # Alternativ: Well-Known Names aus dem Payload lesen
    rtIds: # Einschränkung auf diese Quell-Entitäten
      - 68a2b1c4de73e7b175575402
    rtIdsPath: $.rtIds # Alternativ: Runtime-IDs aus dem Payload lesen
    fieldFilters: # Zusätzliche Filter, mit UND verknüpft
      - attributePath: DataQuality
        operator: GreaterEqualsThan
        comparisonValue: 90
    from: 2026-07-01T00:00:00 # Beginn des aggregierten Zeitraums (UTC)
    fromPath: $.from # Alternativ: Beginn des Zeitraums aus dem Payload lesen
    to: 2026-08-01T00:00:00 # Ende des aggregierten Zeitraums (UTC)
    toPath: $.to # Alternativ: Ende des Zeitraums aus dem Payload lesen
    requireGapFree: true # Nur aggregieren, wenn der Zeitraum lückenlos abgedeckt ist
    expectedInterval: PT15M # Intervall, in dem die Vollständigkeitsprüfung zählt
    maxGapScanRows: 200000 # Obergrenze der Zeilen für die Vollständigkeitsprüfung
    targetPath: $.monthly # Pfad, unter dem die Kennzahlen im Payload abgelegt werden

Parameter​

ParameterTypErforderlichBeschreibung
archiveRtIdstringJaRuntime-ID des Archivs, aus dem gelesen wird. Das Archiv muss aktiviert sein
aggregationsArray von AggregationColumnDtoJaDie zu bildenden Kennzahlen. Mindestens ein Eintrag; dieselbe Spalte darf mehrfach mit unterschiedlichen Funktionen vorkommen
groupByArray von stringNeinSpalten zur Gruppierung, z. B. rtId für eine Zeile je Quell-Entität. Leer ergibt eine einzige Zeile
wellKnownNamesArray von stringNeinSchränkt die Aggregation auf Quell-Entitäten mit diesen Well-Known Names ein
wellKnownNamesPathstringNeinJSONPath-Alternative zu wellKnownNames; akzeptiert einen Einzelwert, ein Array oder einen Multi-Match-Pfad
rtIdsArray von stringNeinSchränkt die Aggregation auf diese Quell-Entitäten ein
rtIdsPathstringNeinJSONPath-Alternative zu rtIds
fieldFiltersArray von FieldFilterWithPathDtoNeinZusätzliche Filter auf die Spalten des Archivs, mit UND verknüpft
fromDatum/ZeitNeinBeginn des aggregierten Zeitraums (UTC). Ohne Wert wird ab Archivbeginn aggregiert
fromPathstringNeinJSONPath zum Beginn des Zeitraums im Payload
toDatum/ZeitNeinEnde des aggregierten Zeitraums (UTC). Ohne Wert wird bis zum Archivende aggregiert
toPathstringNeinJSONPath zum Ende des Zeitraums im Payload
requireGapFreebooleanNeinNur aggregieren, wenn jede Quell-Entität für den gesamten Zeitraum Daten geliefert hat
expectedIntervalZeitspanneNeinIntervall, in dem die Vollständigkeitsprüfung zählt, z. B. PT15M. Standard ist die Periode des Archivs
maxGapScanRowsintegerNeinObergrenze der Zeilen für die Vollständigkeitsprüfung (Standard 200000), muss größer als null sein

Die verfügbaren Filteroperatoren sind bei GetRtEntitiesByType@1 beschrieben.

Aggregationsspalten​

ParameterTypErforderlichBeschreibung
attributePathstringJaDie zu aggregierende Spalte (z. B. Energy, Amount.Value)
functionenumJaDie Aggregationsfunktion
comparisonValuestringNeinFür zustandsbasierte Aggregationen vorgesehen; von den unten genannten Funktionen ignoriert

Unterstützt werden die Funktionen Count, Minimum, Maximum, Average und Sum.

hinweis

TimeWeightedAverage und StateDuration lehnt dieser Node ab: Sie benötigen Metadaten, die der Node nicht tragen kann — einen Vergleichswert bzw. den Fortschreibungspfad eines Raw-Archivs —, und ihre Ergebnisschlüssel folgen anderen Regeln. Verwenden Sie dafür GetQueryById@1 mit einer persistierten Query, die die Aggregation je Spalte definiert.

Vorrang und Zeitzone​

Je Wert gewinnt das Literal gegenüber seiner JSONPath-Variante: from vor fromPath, to vor toPath, wellKnownNames vor wellKnownNamesPath und rtIds vor rtIdsPath.

hinweis

Zeitstempel werden als UTC gelesen. Ein Wert ohne Zeitzonen-Offset (2026-07-01T00:00:00) wird als UTC interpretiert, nicht als Lokalzeit des Adapter-Hosts.

Ein Pfad, der ins Leere zeigt, lässt die Grenze offen und erzeugt eine Warnung; ein vorhandener, aber nicht datumsartiger Wert lässt den Node fehlschlagen.

Spaltennamen​

Die Namen in groupBy, fieldFilters und in den attributePath-Werten folgen demselben Vokabular wie bei GetStreamData@1: Timestamp, WindowStart / WindowEnd (nur bei Archiven mit Fenstern), WellKnownName oder jede vom Archiv deklarierte Spalte. Ein nicht auflösbarer Name lässt den Node fehlschlagen, statt ignoriert zu werden.

vorsicht

Bei einer Aggregation steht mehr auf dem Spiel als beim Lesen: Ein verworfener Filter würde die Kennzahl erhöhen, eine verworfene Gruppierungsspalte alle Gruppen zu einer einzigen Zeile zusammenfallen lassen. Deshalb werden unbekannte Namen abgelehnt und nicht übersprungen.

Ergebnisform​

Das Ergebnis enthält zuerst die Gruppierungsspalten, danach eine Spalte je Kennzahl. Ohne groupBy gibt es genau eine Zeile — auch dann, wenn die Speicherschicht nichts zurückgeliefert hat; die Werte sind in diesem Fall null, statt dass die Zeile fehlt. Ein nachgelagerter Node findet damit immer die erwartete Struktur vor.

Wird derselbe Attributpfad mehrfach aggregiert, wird die Funktion an die Spaltenüberschrift angehängt, da ein reiner Pfadname nicht eindeutig wäre:

Energy (Minimum) | Energy (Maximum)

Vollständigkeitsprüfung​

requireGapFree: true führt die Abdeckungsprüfung von GetStreamData@1 vor der Aggregation aus und nutzt dabei denselben Scan, dieselbe Zeilenobergrenze und denselben Intervall-Rückfall. Fehlen einer Entität Daten, schlägt der Node fehl und benennt die betroffenen Serien mit dem fehlenden Umfang und dem Beginn der ersten Lücke.

Teilergebnisse gibt es nicht: Ein unvollständiger Monat darf keine Kennzahl liefern, die gültig aussieht, aber zu niedrig ist. Die Prüfung setzt beide Zeitgrenzen und ein Archiv mit Fenstern (TimeRange oder Rollup) voraus.

Fallstricke aggregierter Stream Data
  • Sum ist nur bei disjunkten Fenstern korrekt. Identische Fenster kann es nicht geben — die Speicherschicht führt sie zusammen —, überlappende Fenster sind dagegen zulässig und ihr Überlappungsbereich wird doppelt gezählt. requireGapFree schlägt darauf bewusst nicht fehl; es sichert Lückenfreiheit zu, nicht Überschneidungsfreiheit. Wer darauf reagieren muss, wertet hasOverlaps aus dem Lückenbericht von GetStreamData@1 aus.
  • Average ist arithmetisch, nicht zeitgewichtet. Bei gleich langen Fenstern (dem Viertelstundenfall) ist beides identisch; bei variablen Fensterlängen ist ein zeitgewichteter Mittelwert das richtige Mittel — und damit GetQueryById@1.
  • Keine Sortierung, Paginierung oder Zeilenobergrenze. Sie haben für eine Aggregation keine Bedeutung und fehlen deshalb in der Konfiguration, statt vorhanden und wirkungslos zu sein.

Anwendungsbeispiel​

Die im Monat verbrauchte Energie je Zähler aufsummieren und die maximale Datenqualität desselben Zeitraums ausgeben — aber nur, wenn der Monat lückenlos vorliegt:

triggers:
  - type: FromExecutePipelineCommand@1
transformations:
  - type: AggregateStreamData@1
    archiveRtId: 68a1f0c5de73e7b175575401
    aggregations:
      - attributePath: Energy
        function: Sum
      - attributePath: DataQuality
        function: Maximum
    groupBy:
      - rtId
    from: 2026-07-01T00:00:00
    to:   2026-08-01T00:00:00
    requireGapFree: true
    expectedInterval: PT15M
    targetPath: $.monthly
  - type: QueryResultToMarkdownTable@1
    path: $.monthly
    targetPath: $.table

Anwendungsfälle​

  • Abrechnung: Verbrauch eines Abrechnungszeitraums je Zähler ermitteln, abgesichert gegen unvollständige Daten
  • Kennzahlen-Reporting: Monats-, Wochen- oder Schichtsummen, Minima und Maxima bilden
  • Qualitätsüberwachung: Schlechteste oder durchschnittliche Datenqualität eines Zeitraums melden
  • Grenzwertüberwachung: Eine Kennzahl bilden und in einem nachfolgenden Control-Node darauf verzweigen
  • Verdichtung vor der Übertragung: Wenige Kennzahlen statt tausender Rohdatenzeilen versenden

Siehe auch​