Aggregeer data met venstertransformaties in datastroomgrafieken

Een venstertransformatie groepeert binnenkomende berichten en produceert een enkel uitvoerbericht met geaggregeerde waarden wanneer het venster sluit. In plaats van elke meting afzonderlijk door te sturen, kun je statistieken berekenen zoals gemiddelden, minimumwaarden of tellingen en één geconsolideerd resultaat verder doorsturen.

Momenteel kan een venster sluiten op basis van duur, aantal, geheugen of triggercondities.

Zie Het overzicht van gegevensstroomgrafieken voor een overzicht van gegevensstroomgrafieken en hoe transformaties in een pijplijn worden samengesteld.

Opmerking

Voor vensterindeling die niet op duur is gebaseerd, is azureiotoperations/graph-dataflow-window:1.1.0 of hoger vereist.

Transformaties gebruiken een expressietaal om waarden, testvoorwaarden en referentievelden te berekenen. Expressies verwijzen naar invoer op positie, niet op naam: de eerste invoer in de inputs lijst is $1, de tweede is $2, enzovoort. Ingebouwde functies zoals cToF converteren en manipuleren deze waarden.

Voor de volledige lijst van operatoren, functies, datatypes en metadatavelden, zie de referentie Expressions.

Venstertransformaties voegen aggregatiefuncties toe zoals average, min, en max, die alleen beschikbaar zijn in accumulatieregels. Voor de volledige lijst, zie Aggregatiefuncties.

Vereiste voorwaarden

Een exemplaar van Azure IoT-bewerkingen geïmplementeerd in een Kubernetes-cluster. Zie Deploy Azure IoT-bewerkingen voor meer informatie.

Een standaardregistereindpunt dat naar defaultmcr.microsoft.com verwijst wordt automatisch gemaakt tijdens de implementatie.

De Azure CLI voorbeelden in dit artikel gebruiken omgevingsvariabelen zodat je elke waarde één keer kunt instellen en vervolgens de commando's kunt kopiëren en plakken as-is. Als je de Azure IoT-bewerkingen Codespaces-omgeving vanuit de quickstart gebruikt, zijn deze variabelen al voor je ingesteld en kun je deze stap overslaan. Anders stel je de volgende omgevingsvariabelen in je shell voordat je de commando's uitvoert.

De volgende scripts stellen de meest gebruikte omgevingsvariabelen in:

Omgevingsvariabele	Beschrijving
`SUBSCRIPTION_ID`	De ID van het abonnement dat je Azure IoT-bewerkingen-instantie bevat.
`RESOURCE_GROUP`	De naam van de resourcegroep die je Azure IoT-bewerkingen-instantie bevat.
`AIO_INSTANCE_NAME`	De naam van je Azure IoT-bewerkingen instance. Om je instanties op te sommen, voer `az iot ops list -o table`je .
`CLUSTER_NAME`	De naam van de Azure Arc-enabled Kubernetes-cluster die jouw instantie host.
`LOCATION`	De Azure-regio om te gebruiken voor nieuwe bronnen, bijvoorbeeld `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>"

Je hoeft alleen de variabelen in te stellen die dit artikel gebruikt. Dit artikel kan extra omgevingsvariabelen gebruiken voor de bronnamen die je kiest. Het artikel legt uit hoe je ze op de plek kunt plaatsen waar ze worden geïntroduceerd.

Schaalbaarheidsbeperking voor statusbehoudende grafen

Belangrijk

Raam- en gashendeltransformaties zijn stateful. Elke instantie behoudt zijn eigen staat en instanties delen die toestand niet met elkaar. Wanneer het aantal instanties in het dataflowprofiel groter is dan één, verdelen gedeelde abonnementen berichten over instanties , zodat elke instantie slechts een deelverzameling van de berichten ziet. Een windowtransform berekent vervolgens aggregaties zoals gemiddelden, sommen en tellingen over een gedeeltelijke dataset, en een throttle-transformatie handhaaft de geconfigureerde snelheidslimiet onafhankelijk in elke instantie in plaats van over de hele pijplijn.

Stel het aantal instanties van het dataflowprofiel in op 1 voor elke datastroomgrafiek die een window- of throttle-transformatie gebruikt. Stateless dataflow-grafieken die alleen map-, filter-, branch- en concatenate-transformaties gebruiken, kunnen veilig hogere instance-tellingen gebruiken om de doorvoersnelheid te verhogen.

Wanneer u een venstertransformatie gebruikt

Gebruik een venstertransformatie wanneer u sensorgegevens met een hoge frequentie ontvangt en het volume wilt verminderen voordat u het downstream verzendt. Veelvoorkomende scenario's zijn onder andere:

Rekengemiddelden: een temperatuursensor publiceert elke seconde, maar uw cloudtoepassing heeft slechts een gemiddelde van 30 seconden nodig.
Extremen bijhouden: u wilt de minimale en maximale drukmetingen gedurende elk interval van één minuut.
Aantal gebeurtenissen: U moet weten hoeveel doorgangsgebeurtenissen de afgelopen vijf minuten hebben plaatsgevonden.
Maak productiebatches: Je wilt statistieken berekenen voor elke batch van vaste grootte, zoals elke 100 pakketten die van een vullijn komen.
Reageren op toestandsveranderingen: Je wilt weten wanneer een bedieningssignaal verandert, bijvoorbeeld wanneer een mixer verandert van running naar draining.

Hoe de venstertransformatie werkt

De venstertransformatie heeft twee interne stappen die op volgorde zijn verbonden:

Window: Bewaart berichten in een buffer totdat een van de geconfigureerde sluitvoorwaarden wordt geactiveerd.
Verzamelen: Past uw aggregatieregels toe wanneer het venster wordt gesloten. De transformatie reduceert alle berichten in het venster tot één enkel uitvoerbericht.

Opmerking

Een venstertransformatie moet ten minste één sluitingsvoorwaarde configureren: delay, , count, memory, of triggers.

Configureer de sluitingsvoorwaarden van het venster

Vanaf versie 1.1.0, voegt de venstertransformatie drie configuratiesleutels toe naast de bestaande delay sleutel:

Configuratiesleutel	Raamtype	Purpose
`delay`	Duurgebaseerd venster	Sluit het venster na een vaste periode.
`count`	Tel-gebaseerd venster	Sluit het venster na een vast aantal berichten.
`memory`	Geheugengebaseerd venster	Sluit het venster wanneer de grootte van de gebufferde payload een limiet bereikt.
`triggers`	Door triggers aangestuurd venster	Sluit het venster wanneer een aangepaste expressie wordt geëvalueerd als `true`.

Duurgebaseerd venster

Gebruik de delay configuratie om het venster na een vaste duur te sluiten. Deze instelling bepaalt hoe lang elk tumbling-venster duurt.

Opmerking

Met de vertragingsstap worden berichttijdstempels uitgelijnd op venstergrenzen. Als een bericht 7 seconden na een venster van 10 seconden binnenkomt, behoort het tot de 10-secondengrens.

Opmerking

Als je het niet geeft delay, gebruikt het venster een standaard timeout van 60 seconden als veiligheidsklep.

Stel in de configuratie van de venstertransformatie de duur van het venster in seconden in. Stel deze bijvoorbeeld in op 30 voor een 30-seconden "tumbling window".

De CLI past de hele grafiek uit één configuratiebestand toe. Voeg deze configuratie toe aan de venstertransformatie-nodes configuration in je graph.json bestand. Pas het vervolgens toe met az iot ops dataflowgraph apply.

De regels zijn een JSON-object:

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

Deze regels gaan in het value veld als een escape-tekenreeks:

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

Tip

Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Vastgoed	Typ	Beschrijving
`type`	tekenreeks	Moet `"duration"`zijn.
`delaySeconds`	uint64	Aantal seconden voordat het venster sluit. Moet groter zijn dan 0.

Tel-gebaseerd venster

Gebruik de count configuratie om het venster te sluiten na een vast aantal berichten.

In de configuratie van de window transform zet u Aantal berichten op 5 en stel u het gedrag van grensberichten in op messageInCurrent.

De CLI past de hele grafiek toe vanuit één configuratiebestand, dus voeg deze configuratie toe aan de venstertransformatie-nodes configuration in je graph.json bestand. Pas het vervolgens toe met az iot ops dataflowgraph apply.

De regels zijn een JSON-object:

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

Voeg deze regels toe in het value-veld als een geëscapete tekenreeks:

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

Tip

Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Vastgoed	Typ	Beschrijving
`type`	tekenreeks	Moet `"messageCount"`zijn.
`maxMessageCount`	uint64	Aantal berichten dat moet worden gebufferd voordat het venster sluit. Moet groter zijn dan 0.
`boundaryMessage`	tekenreeks	Of het bericht dat het venster sluit in het huidige venster blijft (`messageInCurrent`) of het volgende venster start (`messageInNext`).

Geheugengebaseerd venster

Gebruik de memory configuratie om het venster te sluiten wanneer de gebufferde payloadgrootte een limiet bereikt.

Stel in de configuratie van de venstertransformatie Buffergrootte in op 1048576 bytes en stel het gedrag van grensberichten in op messageInNext.

De regels zijn een JSON-object:

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

Voeg deze regels toe in het value-veld als een geëscapete tekenreeks:

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

Tip

Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Vastgoed	Typ	Beschrijving
`type`	tekenreeks	Moet `"bufferSize"`zijn.
`maxBufferBytes`	uint64	Maximale cumulatieve payload-bytes voordat het venster sluit. Moet groter zijn dan 0.
`boundaryMessage`	tekenreeks	Of het bericht dat het venster sluit in het huidige venster blijft (`messageInCurrent`) of het volgende venster start (`messageInNext`).

Door triggers aangestuurd venster

Gebruik de triggers configuratie wanneer het venster zou moeten sluiten op basis van de inhoud van het bericht of de loopstatus binnen het huidige venster.

Voeg in de configuratie van de venstertransformatie een triggerregel toe met invoerveld temperature, expressie running_sum($1) + $1 > 100, en grensberichtgedrag messageInCurrent.

De regels zijn een JSON-object:

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

Voeg deze regels toe in het value-veld als een geëscapete tekenreeks:

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

Tip

Om de ge-escapete tekenreeks te genereren, sla je de regels op in een bestand zoals rules.json, voer je jq -c . rules.json uit en plak je de eenregelige uitvoer in het veld value.

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Vastgoed	Verplicht	Beschrijving
`type`	Ja	Moet `"expression"`zijn.
`rules`	Ja	Array van triggerregels. Regels worden sequentieel per bericht geëvalueerd; De eerste matchingregel sluit het venster.
`datasets`	No	Optionele state-store datasets die naar de state store verwijzen.

Elke triggerregel ondersteunt deze velden:

Vastgoed	Verplicht	Beschrijving
`inputs`	Ja	Reeks van verwijzingen naar invoervelden. De uitdrukking bindt aan `$1`, `$2`, enzovoort.
`trigger`	Ja	Booleaanse uitdrukking die het venster sluit wanneer het evalueert naar `true`.
`boundaryMessage`	Ja	Of het bericht dat het venster sluit in het huidige venster blijft (`messageInCurrent`) of het volgende venster start (`messageInNext`).

Het inputs veld ondersteunt dezelfde invoersyntaxis die elders in datastroomgrafieken wordt gebruikt, waaronder gewone velden, ?? standaarden, ? $last, , $context(key).fielden $metadata.*. Voor meer details over het gebruik $context(key), zie Verrijken met externe data.

Triggerexpressies kunnen de reguliere grafiekexpressiefuncties gebruiken en de volgende functies voor de actieve status, die worden gereset wanneer het venster wordt gesloten:

Function	Beschrijving
`running_sum($1)`	Cumulatieve som van `$1` over eerdere berichten in het huidige venster.
`running_avg($1)`	Cumulatief gemiddelde van `$1` eerdere berichten.
`running_min($1)`	Minimale waarde van `$1` die in eerdere berichten is gezien. Retourneert `$1` bij het eerste bericht (het minimum van één element is dat element zelf).
`running_max($1)`	Maximale waarde van `$1` in eerdere berichten. Retourneert `$1` bij het eerste bericht (het maximum van een enkel element is het element zelf).
`running_count($1)`	Aantal berichten waar `$1` aanwezig was.
`running_count()`	Totaal aantal berichten (geen veldfilter).
`first($1)`	Eerste niet-lege waarde van `$1` in het huidige venster. Keert terug `$1` bij het eerste bericht.
`changed($1)`	`true` als `$1` verschilt van de waarde in het vorige bericht. `false` op het eerste bericht van een venster (geen eerdere waarde om mee te vergelijken).
`prev($1)`	Meest recente niet-lege waarde van `$1` uit een eerder bericht in het huidige venster. Berichten waarbij `$1` Leeg was, worden overgeslagen (de opgeslagen waarde wordt niet overschreven). Retourneert `$1` voor het eerste bericht in een venster.

Opmerking

running_sum($1) en vergelijkbare functies geven waarden terug van eerder verwerkte berichten. Voor het huidige bericht, gebruik $1.

Voorbeelden van triggerregels

Gebruik deze voorbeelden om gemeenschappelijke inputs patronen en patronen trigger te zien in een compleet triggerconfiguratieobject:

Dit voorbeeld laat een gewone triggerexpressie zien. Het venster wordt gesloten wanneer de huidige temperature groter is dan 80.

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

Dit voorbeeld toont een triggeruitdrukking die running_sum($1) + $1 eerdere berichten in het huidige venster combineert met het huidige bericht, en vervolgens het venster sluit wanneer de drempel wordt overschreden.

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

Dit voorbeeld toont nulveilige invoerbehandeling met temperature ?? 0, plus messageInNext om het grensbericht in het volgende venster te plaatsen.

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

Dit voorbeeld toont een metadata-gebaseerde trigger waarbij het venster sluit voor een specifieke onderwerpwaarde uit $metadata.topic.

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

Dit voorbeeld toont triggerregels met datasetverrijking: het koppelt het bericht factoryId aan een state-store-rij, leest shiftId uit $context(factory).shiftId, en sluit het venster wanneer die verschuivingswaarde verandert (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 dit voorbeeld wordt verwacht dat de state store-dataset die wordt weergegeven door factory velden bevat zoals factoryId en shiftId.

Grensgedrag

De boundaryMessage instelling bepaalt wat er gebeurt met het bericht waardoor een tel-, geheugen- of triggervenster is gesloten:

messageInCurrent: neem het grensbericht op in het sluitvenster.
messageInNext: sluit eerst het huidige venster, en start dan het volgende venster met het grensbericht.

Als messageInNext bij het eerste bericht in een nieuw venster wordt geactiveerd, wordt het sluiten onderdrukt zodat er geen leeg venster wordt gegenereerd.

Opmerking

Het venster op basis van duur maakt geen gebruik van boundaryMessage. Duurgrenzen zijn tijdsgebaseerd, niet berichtgebaseerd, dus er is geen grensbericht om in het huidige of volgende venster te plaatsen.

Combineer sluitingsvoorwaarden

Je kunt duur, aantal, geheugen en triggercondities combineren in dezelfde grafiek.

De duur wordt tijdgestuurd en wordt beoordeeld door de timer.
Voor elk binnenkomend bericht worden berichtgestuurde condities geëvalueerd in deze volgorde: Memory > Count > Trigger.
Binnen triggers.rulesworden de regels sequentieel geëvalueerd en wint de eerste matchingregel.

De evaluatievolgorde in berichtgestuurde voorwaarden is van belang voor de accumulatieresultaten wanneer een bericht aan meerdere voorwaarden tegelijk voldoet. Bijvoorbeeld, als memorymessageInCurrent gebruikt en countmessageInNext gebruikt, volgt er na de geheugenconfiguratie een bericht dat aan beide voorwaarden voldoet. Het bericht blijft in het huidige venster en draagt bij aan de accumulatie-output van dat venster.

Accumulatieregels definiëren

Elke accumulatieregel geeft aan hoe u een venster met berichten in één uitvoerwaarde kunt verminderen. De configuratiesleutel is rules.

Voeg in de windowtransformatieconfiguratie een accumulatieregel toe met input temperature, output avgTemperature, en aggregatiefunctie average($1).

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg dit toe aan het venstertransformatieknooppunt configuration in uw graph.json bestand en pas deze toe met az iot ops dataflowgraph apply.

De regels zijn een JSON-object:

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

Deze regels gaan in het value veld als een escape-tekenreeks:

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

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

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Vastgoed	Verplicht	Beschrijving
`inputs`	Ja	Lijst met veldpaden die moeten worden gelezen uit elk binnenkomend bericht.
`output`	Ja	Veldpad voor het geaggregeerde resultaat. Elke regel moet een unieke uitvoer hebben.
`expression`	Ja	Formule die invoerwaarden in het venster vermindert tot één scalaire waarde. Moet ten minste één aggregatiefunctie bevatten.
`description`	No	Leesbare beschrijving.

In tegenstelling tot kaartregels is expressionvereist voor elke accumulatieregel. Alleen $1 gebruiken is niet geldig omdat deze verwijst naar een verzameling waarden, niet naar één scalaire waarde. U moet deze verpakken in een aggregatiefunctie zoals average($1).

Aggregatiefuncties

Function	Retouren	Gedrag van leeg venster
`average`	Gemiddelde van numerieke waarden	Fout
`sum`	Som van numerieke waarden	0,0
`min`	Minimale numerieke waarde	Fout
`max`	Maximumaantal numerieke waarden	Fout
`count`	Aantal berichten waarin het veld bestaat	0
`first`	Eerste waarde in het venster	Fout
`last`	Laatste waarde in het venster	Fout

Elke functie gebruikt één positionele variabele als argument ($1 voor de eerste invoer, $2 voor de tweede, enzovoort).

Niet-numerieke waarden: De functies average, sum, min en max slaan niet-numerieke waarden stilzwijgend over.

Functies op basis van aanwezigheid: count, firsten last werken op aanwezigheid van velden, ongeacht het waardetype.

Aggregaties combineren

Combineer meerdere aggregatiefuncties in één expressie:

Voeg een regel toe met invoer temperature en humidityexpressie average($1) + max($2).

De CLI past de hele grafiek uit één configuratiebestand toe, dus voeg deze toe aan de bijbehorende plaats in uw graph.json bestand en pas deze toe met 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)'
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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

Als u een geaggregeerde waarde wilt converteren, past u de conversiefunctie buiten de aggregatie toe. Converteert bijvoorbeeld cToF(average($1)) de gemiddelde temperatuur naar Fahrenheit.

Elke aggregatiefunctie moet rechtstreeks verwijzen naar één positionele variabele. average($1) + max($2) is geldig, maar average($1 + $2) niet.

Verschillen tussen kaartregels

Vermogen	Kaartregels	Accumulatieregels
Expressie vereist	No	Ja
Jokertekeninvoer	Ondersteund	Niet ondersteund
`$metadata` Toegang	Ondersteund	Niet ondersteund
`$context` Verrijking	Ondersteund	Niet ondersteund
`? $last` richtlijn	Ondersteund	Niet ondersteund
Uitvoerinhoudstype	Komt overeen met invoer	Altijd `application/json`

Voorbeeld van volledige configuratie

Dit voorbeeld toont een volledige vensterconfiguratie die het venster sluit na 30 seconden, 5 berichten, 1.048.576 gebufferde bytes, of wanneer running_sum($1) + $1 > 100. Het voorbeeld stelt de boundaryMessage waarde in voor messageInCurrent de laatste drie voorwaarden, en het venster berekent temperatuurstatistieken wanneer het sluit.

Welke voorwaarde het venster sluit hangt af van de berichttiming, het aantal, de grootte van de payload en de inhoud. De volgende voorbeelden tonen de resulterende output voor elke sluitingsconditie.

Duur eindigt

Als er geen andere voorwaarde eerst wordt geactiveerd en het venster 30 seconden na ontvangst van deze drie berichten bereikt:

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

Het uitvoerbericht is:

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

Aantal sluiten

Als het venster deze vijf berichten ontvangt voordat enige andere voorwaarde optreedt:

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

Het uitvoerbericht is:

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

Geheugen wordt gesloten

Als de omvang van de gebufferde payload 1.048.576 bytes bereikt voordat een andere voorwaarde wordt vervuld, bijvoorbeeld na deze twee grote berichten:

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

Het uitvoerbericht is:

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

Trigger gesloten

Als de trigger-expressie running_sum($1) + $1 > 100 vóór elke andere voorwaarde wordt afgevuurd, bijvoorbeeld na deze drie berichten:

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

Het uitvoerbericht is:

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

Maak in de bewerkingservaring een gegevensstroomgrafiek met een venstertransformatie:

Voeg een bron toe waaruit telemetry/temperaturewordt gelezen.
Een venstertransformatie toevoegen. Configureer een duurperiode van 30 seconden, een limiet van 5 berichten, een bufferlimiet van 1.048.576 bytes, en een triggerregel op temperature met expressie running_sum($1) + $1 > 100. Voor de tel-, geheugen- en triggervoorwaarden stel je het gedrag van grensberichten in op messageInCurrent. Voeg opbouwregels toe voor gemiddelde, minimum, maximaal, aantal en bereik op het temperature veld.
Voeg een bestemming toe om naar telemetry/aggregated te verzenden.

De Azure CLI past een gegevensstroomgrafiek van één JSON-configuratiebestand toe. Maak een graph.json bestand met de grafiekeigenschappen. In het graph.json bestand sla je de regels van elke transformatie in het value veld op als een ontsnapte JSON-string. Zie de procedure voor dat transformatietype voor de leesbare vorm van de regels van elke transformatie.

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

Pas het configuratiebestand toe.

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' } }
    ]
  }
}

Belangrijk

Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.

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 }

Volgende stappen

Feedback

Is deze pagina nuttig?

Last updated on 2026-08-14

Aggregeer data met venstertransformaties in datastroomgrafieken

Vereiste voorwaarden

Schaalbaarheidsbeperking voor statusbehoudende grafen

Wanneer u een venstertransformatie gebruikt

Hoe de venstertransformatie werkt

Configureer de sluitingsvoorwaarden van het venster

Duurgebaseerd venster

Tel-gebaseerd venster

Geheugengebaseerd venster

Door triggers aangestuurd venster

Voorbeelden van triggerregels

Grensgedrag

Combineer sluitingsvoorwaarden

Accumulatieregels definiëren

Aggregatiefuncties

Aggregaties combineren

Verschillen tussen kaartregels

Voorbeeld van volledige configuratie

Duur eindigt

Aantal sluiten

Geheugen wordt gesloten

Trigger gesloten

Volgende stappen

Feedback

Aanvullende resources