Een gegevensstroomgrafiek is een samenstelbare verwerkingspijplijn waarmee gegevens worden getransformeerd wanneer deze worden verplaatst tussen bronnen en bestemmingen. Een standaardgegevensstroom volgt een vaste verrijking, filter, kaartvolgorde. Met een gegevensstroomgrafiek kunt u transformaties in elke willekeurige volgorde ketenen, vertakken naar parallelle paden en gegevens samenvoegen over tijdvensters.
In dit artikel wordt stapsgewijs uitgelegd hoe u een gegevensstroomgrafiek maakt. Voor een overzicht van gegevensstroomgrafieken en de beschikbare transformaties, zie Data flow graphs overview.
Gegevensstroomgrafieken ondersteunen momenteel alleen MQTT-, Kafka- en OpenTelemetry-eindpunten. Ze ondersteunen geen andere endpointtypes zoals Data Lake, Microsoft Fabric OneLake, Azure Data Explorer en Local Storage.
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.
Vereiste voorwaarden
De Azure CLI geïnstalleerd op uw ontwikkelcomputer. Controleer Beschikbare Azure CLI-extensies voor de minimaal vereiste versie om de azure-iot-ops-extensie te gebruiken. Gebruik az --version om uw versie te controleren en az upgrade om indien nodig bij te werken. Zie De Azure CLI installeren voor meer informatie.
De Azure IoT-bewerkingen-extensie voor de Azure CLI. Gebruik de volgende opdracht om de extensie toe te voegen of bij te werken naar de nieuwste versie:
az extension add --upgrade --name azure-iot-ops
Azure IoT-bewerkingen versie 1.2 of hoger.
Een gegevensstroomprofiel. U kunt het standaardprofiel gebruiken.
Een gegevensstroomeindpunt voor uw bron en bestemming. Het standaardeindpunt van de MQTT-broker is geschikt om mee te beginnen.
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 tableje . |
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. |
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.
Dit artikel gebruikt ook de GRAPH_NAME en PROFILE omgevingsvariabelen voor de datastroomgrafiek en profielnamen. Stel elk commando in voordat je de bijbehorende commando's uitvoert.
Een gegevensstroomgrafiek maken
Een gegevensstroomgrafiek bevat drie typen elementen: bronnen die gegevens binnenbrengen, transformeren die het verwerken en bestemmingen die het verzenden. Verbind ze in de volgorde waarin u gegevens wilt laten stromen.
In de operationele ervaring ga je naar je Azure IoT-bewerkingen-instantie.
Selecteer Gegevensstroomgrafiek>Gegevensstroomgrafiek maken.
Voer een naam in voor de gegevensstroomgrafiek en selecteer een gegevensstroomprofiel. Het standaardprofiel is standaard geselecteerd.
Bouw uw pijplijn door elementen toe te voegen aan het canvas:
Een bron toevoegen: selecteer het broneindpunt en configureer de onderwerpen waarop u zich wilt abonneren voor binnenkomende berichten.
Transformaties toevoegen: Selecteer een of meer transformaties om de gegevens te verwerken. Beschikbare transformaties zijn map, filter, branch, concatenate, window en throttle. Zie het overzicht van gegevensstroomgrafieken voor meer informatie over elk transformatietype.
Een bestemming toevoegen: selecteer het doeleindpunt en configureer het onderwerp of pad waar verwerkte gegevens naartoe moeten worden verzonden.
Verbind de elementen in de volgorde waarin u gegevens wilt laten stromen.
Selecteer Opslaan om de gegevensstroomgrafiek te implementeren.
De Azure CLI gebruikt een dataflowgrafiek uit één JSON-configuratiebestand dat alle knooppunten en verbindingen bevat. Gebruik az iot ops dataflowgraph apply om de grafiek te maken of te vervangen. Het volgende voorbeeld leest temperatuurgegevens, zet deze om naar Fahrenheit en stuurt deze naar een bestemmingsonderwerp.
Maak een graph.json bestand met de eigenschappen van de gegevensstroomgrafiek. 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": "convert",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
}
]
}
},
{
"nodeType": "Destination",
"name": "output",
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "telemetry/converted"
}
}
],
"nodeConnections": [
{
"from": {
"name": "sensors"
},
"to": {
"name": "convert"
}
},
{
"from": {
"name": "convert"
},
"to": {
"name": "output"
}
}
]
}
Pas het configuratiebestand toe.
az iot ops dataflowgraph apply \
--name temperature-processing \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP \
--config-file graph.json
De grafiek gebruikt het default datastroomprofiel. Als u een ander profiel wilt gebruiken, voegt u dit toe --profile <PROFILE_NAME>.
Maak een Bicep-bestand .bicep met de volgende structuur. In dit voorbeeld wordt een gegevensstroomgrafiek gemaakt waarmee temperatuurgegevens worden gelezen, geconverteerd naar Fahrenheit en naar een doelonderwerp worden verzonden.
param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
name: customLocationName
}
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
parent: aioInstance
name: 'default'
}
resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
parent: defaultDataflowProfile
name: 'temperature-processing'
extendedLocation: {
name: customLocation.id
type: 'CustomLocation'
}
properties: {
profileRef: 'default'
mode: 'Enabled'
nodes: [
{
nodeType: 'Source'
name: 'sensors'
sourceSettings: {
endpointRef: 'default'
dataSources: [
'telemetry/temperature'
]
}
}
{
nodeType: 'Graph'
name: 'convert'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
}
]
}
}
{
nodeType: 'Destination'
name: 'output'
destinationSettings: {
endpointRef: 'default'
dataDestination: 'telemetry/converted'
}
}
]
nodeConnections: [
{
from: { name: 'sensors' }
to: { name: 'convert' }
}
{
from: { name: 'convert' }
to: { name: 'output' }
}
]
}
}
Het Bicep-bestand implementeren:
az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
Maak een Kubernetes-manifestbestand .yaml met de volgende structuur. In dit voorbeeld wordt een gegevensstroomgrafiek gemaakt waarmee temperatuurgegevens worden gelezen, geconverteerd naar Fahrenheit en naar een doelonderwerp worden verzonden.
apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
name: temperature-processing
namespace: azure-iot-operations
spec:
profileRef: default
nodes:
- nodeType: Source
name: sensors
sourceSettings:
endpointRef: default
dataSources:
- telemetry/temperature
- nodeType: Graph
name: convert
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{
"inputs": ["*"],
"output": "*"
},
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
- nodeType: Destination
name: output
destinationSettings:
endpointRef: default
dataDestination: telemetry/converted
nodeConnections:
- from:
name: sensors
to:
name: convert
- from:
name: convert
to:
name: output
Pas het manifest toe:
kubectl apply -f main.yaml
De bron definieert waar gegevens de pijplijn binnenkomen. Geef een eindpuntreferentie en een of meer onderwerpen op.
Selecteer in de grafiekeditor voor gegevensstromen het bronelement en configureer het volgende:
| Configuratie |
Beschrijving |
|
Eindpunt |
Het eindpunt van de gegevensstroom dat moet worden gebruikt. Selecteer de standaardwaarde voor de lokale MQTT-broker. |
| Onderwerpen |
Een of meer onderwerpen waarop u zich kunt abonneren voor binnenkomende berichten. |
De CLI past de hele grafiek tegelijk toe, dus configureer de bron als een Source knooppunt in uw graph.json configuratiebestand en voer vervolgens het volgende uit az iot ops dataflowgraph apply:
{
"nodeType": "Source",
"name": "sensors",
"sourceSettings": {
"endpointRef": "default",
"dataSources": [
"telemetry/temperature",
"telemetry/humidity"
]
}
}
{
nodeType: 'Source'
name: 'sensors'
sourceSettings: {
endpointRef: 'default'
dataSources: [
'telemetry/temperature'
'telemetry/humidity'
]
}
}
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
- nodeType: Source
name: sensors
sourceSettings:
endpointRef: default
dataSources:
- telemetry/temperature
- telemetry/humidity
Transformeert procesgegevens tussen de bron en het doel. Elke transformatie verwijst naar een ingebouwd artefact en heeft configuratieregels.
De beschikbare ingebouwde transformaties zijn:
| Transformeren |
Artifact |
Beschrijving |
|
Kaart |
azureiotoperations/graph-dataflow-map:1.0.0 |
Hernoem, herstructureer, bereken en kopieer velden. |
|
Filteren |
azureiotoperations/graph-dataflow-filter:1.0.0 |
Berichten verwijderen die overeenkomen met een voorwaarde. |
|
Filiaal |
azureiotoperations/graph-dataflow-branch:1.0.0 |
Routeer elk bericht naar een true of false pad op basis van een voorwaarde. |
|
Concatenate |
azureiotoperations/graph-dataflow-concatenate:1.0.0 |
Voeg twee of meer paden weer samen in één pad. |
|
Venster |
azureiotoperations/graph-dataflow-window:1.0.0 |
Verzamel berichten gedurende een tijdsinterval en aggregeren. |
|
Throttle |
azureiotoperations/graph-dataflow-throttle:1.0.0 |
Beperk de berichtsnelheid per MQTT-themapatroon. |
Zie Verrijken met externe gegevens voor meer informatie over het verrijken van berichten met externe gegevens.
Selecteer transformatie toevoegen in de grafiekeditor voor gegevensstromen en kies het transformatietype. Configureer de regels in de visual-editor.
Elke transformatie is een knooppunt waarbij nodeType is ingesteld op Graph in uw graph.json-configuratiebestand. De regels van de transformatie zijn een JSON-object, zoals deze kaart die temperatuur converteert naar Fahrenheit:
{
"map": [
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
De configuration eigenschap neemt deze regels als een string, dus ontsnap je aan de regels JSON en plaats het in het value veld. Pas de volledige grafiek toe met az iot ops dataflowgraph apply:
{
"nodeType": "Graph",
"name": "convert",
"graphSettings": {
"registryEndpointRef": "default",
"artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
"configuration": [
{
"key": "rules",
"value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
}
]
}
}
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.
Elke transformatie is een knooppunt met nodeType: 'Graph'. De configuration eigenschap geeft regels door als een JSON-tekenreeks:
{
nodeType: 'Graph'
name: 'convert'
graphSettings: {
registryEndpointRef: 'default'
artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
configuration: [
{
key: 'rules'
value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
}
]
}
}
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
Elke transformatie is een knooppunt met nodeType: Graph. De configuration eigenschap geeft regels door als een JSON-tekenreeks:
- nodeType: Graph
name: convert
graphSettings:
registryEndpointRef: default
artifact: azureiotoperations/graph-dataflow-map:1.0.0
configuration:
- key: rules
value: |
{
"map": [
{
"inputs": ["temperature"],
"output": "temperature_f",
"expression": "cToF($1)"
}
]
}
Ketting een willekeurig aantal transformaties aan. Verbind ze in de nodeConnections sectie in de volgorde waarin u gegevens wilt laten stromen:
Sleep de verbindingen tussen transformaties op het canvas om de volgorde van verwerking te bepalen.
Definieer de verwerkingsvolgorde in de nodeConnections sectie van uw graph.json configuratiebestand:
"nodeConnections": [
{
"from": {
"name": "sensors"
},
"to": {
"name": "remove-bad-data"
}
},
{
"from": {
"name": "remove-bad-data"
},
"to": {
"name": "convert"
}
},
{
"from": {
"name": "convert"
},
"to": {
"name": "output"
}
}
]
nodeConnections: [
{ from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
{ from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
{ from: { name: 'convert' }, to: { name: 'output' } }
]
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
nodeConnections:
- from: { name: sensors }
to: { name: remove-bad-data }
- from: { name: remove-bad-data }
to: { name: convert }
- from: { name: convert }
to: { name: output }
De bestemming bepaalt waar verwerkte data naartoe gaat. Geef een eindpuntreferentie en een onderwerp of pad op.
Selecteer het doelelement en configureer het volgende:
| Configuratie |
Beschrijving |
|
Eindpunt |
Het eindpunt van de gegevensstroom om gegevens naar te verzenden. |
|
onderwerp |
Het onderwerp of pad van het publiceren van verwerkte gegevens. |
Configureer de bestemming als een Destination knooppunt in uw graph.json configuratiebestand en pas vervolgens de volledige grafiek toe met az iot ops dataflowgraph apply:
{
"nodeType": "Destination",
"name": "output",
"destinationSettings": {
"endpointRef": "default",
"dataDestination": "telemetry/processed"
}
}
{
nodeType: 'Destination'
name: 'output'
destinationSettings: {
endpointRef: 'default'
dataDestination: 'telemetry/processed'
}
}
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
- nodeType: Destination
name: output
destinationSettings:
endpointRef: default
dataDestination: telemetry/processed
Zie Berichten doorsturen naar verschillende onderwerpen voor dynamische onderwerproutering op basis van berichtinhoud.
Controleren of de gegevensstroomgrafiek werkt
Nadat u een gegevensstroomgrafiek hebt geïmplementeerd, controleert u of deze actief is:
Selecteer in de operationele ervaring je datastroomgrafiek om de status ervan te bekijken. Een gezonde grafiek toont een Running-status.
Gebruik az iot ops dataflowgraph show om de details van een grafiek te bekijken:
az iot ops dataflowgraph show \
--name temperature-processing \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP
Als u alle gegevensstroomgrafieken wilt weergeven die zijn gekoppeld aan een profiel, gebruikt u az iot ops dataflowgraph list:
az iot ops dataflowgraph list \
--instance $AIO_INSTANCE_NAME \
--resource-group $RESOURCE_GROUP
Controleer de status van de DataflowGraph resource:
az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE
Belangrijk
Het gebruik van Kubernetes-implementatiemanifesten wordt niet ondersteund in productieomgevingen en mag alleen worden gebruikt voor foutopsporing en testen.
kubectl get dataflowgraph temperature-processing -n azure-iot-operations
Controleer de podlogboeken op eventuele fouten:
kubectl logs -l app=dataflow -n azure-iot-operations --tail=50
Verwante onderwerpen
