Creare un grafo del flusso di dati nelle operazioni IoT di Azure

Un grafico del flusso di dati è una pipeline di elaborazione componibile che trasforma i dati mentre si sposta tra origini e destinazioni. Un flusso di dati standard segue una sequenza fissa di arricchimento, filtro, mappa. Un grafico del flusso di dati consente di concatenare le trasformazioni in qualsiasi ordine, diramazione in percorsi paralleli e di aggregare i dati nelle finestre temporali.

Questo articolo illustra in dettaglio la creazione di un grafo del flusso di dati. Per una panoramica dei grafici del flusso di dati e delle trasformazioni disponibili, vedere Panoramica dei grafici del flusso di dati.

I grafici del flusso di dati supportano attualmente solo endpoint MQTT, Kafka e OpenTelemetry. Non supportano altri tipi di endpoint come Data Lake, Microsoft Fabric OneLake, Esplora dati di Azure e Local Storage.

Le trasformazioni utilizzano un linguaggio di espressione per calcolare valori, condizioni di test e campi di riferimento. Le espressioni si riferiscono agli input per posizione, non per nome: il primo input nella inputs lista è $1, il secondo è $2, e così via. Funzioni integrate come cToF convertono e manipolano tali valori.

Per l'elenco completo di operatori, funzioni, tipi di dati e campi di metadati, consulta il riferimento Expressions.

Prerequisiti

Istanza di Operazioni di Azure IoT distribuita in un cluster Kubernetes. Per altre informazioni, vedere Deploy Operazioni di Azure IoT.

Il interfaccia della riga di comando di Azure installato nel computer di sviluppo. Controlla le estensioni interfaccia della riga di comando di Azure disponibili per la versione minima richiesta per utilizzare l'estensione azure-iot-ops. Usare az --version per controllare la versione e az upgrade per aggiornarla, se necessario. Per altre informazioni, vedere Installare il interfaccia della riga di comando di Azure.
Estensione Operazioni di Azure IoT per il interfaccia della riga di comando di Azure. Usare il comando seguente per aggiungere l'estensione o aggiornarla alla versione più recente:
```
az extension add --upgrade --name azure-iot-ops
```

Operazioni di Azure IoT versione 1.2 o successiva.
Profilo del flusso di dati. È possibile usare il profilo predefinito.
Endpoint del flusso di dati per l'origine e la destinazione. L'endpoint broker MQTT predefinito funziona per iniziare.

Gli esempi interfaccia della riga di comando di Azure in questo articolo usano variabili di ambiente così puoi impostare ogni valore una volta e poi copiare e incollare i comandi as-is. Se stai usando l'ambiente Operazioni di Azure IoT Codespaces dal quickstart, queste variabili sono già impostate per te e puoi saltare questo passaggio. Altrimenti, imposta le seguenti variabili di ambiente nella tua shell prima di eseguire i comandi.

I seguenti script impostano le variabili di ambiente più comunemente utilizzate:

Variabile di ambiente	Descrizione
`SUBSCRIPTION_ID`	L'ID dell'abbonamento che contiene la tua istanza Operazioni di Azure IoT.
`RESOURCE_GROUP`	Il nome del gruppo di risorse che contiene la tua istanza Operazioni di Azure IoT.
`AIO_INSTANCE_NAME`	Il nome della tua istanza Operazioni di Azure IoT. Per elencare le tue istanze, esegui `az iot ops list -o table`.
`CLUSTER_NAME`	Il nome del cluster Kubernetes abilitato Azure Arc che ospita la tua istanza.
`LOCATION`	La regione Azure da utilizzare per nuove risorse, ad esempio `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>"

Devi solo impostare le variabili utilizzate in questo articolo. Questo articolo potrebbe utilizzare variabili ambientali aggiuntive per i nomi delle risorse che scegli. L'articolo spiega come posizionarli dove vengono introdotti.

Questo articolo utilizza anche le GRAPH_NAME variabili e PROFILE ambiente per il grafico del flusso dati e i nomi dei profili. Imposta ognuno prima di eseguire i comandi correlati.

Creare un grafo del flusso di dati

Un grafico del flusso di dati contiene tre tipi di elementi: origini in cui vengono inseriti i dati, trasformati e destinazioni che lo inviano. Connetterli nell'ordine in cui si desidera che i dati vengano trasmessi.

Nell'esperienza operativa, vai alla tua istanza Operazioni di Azure IoT.
Selezionare Grafico flusso di dati>Crea grafico flusso di dati.
Immettere un nome per il grafico del flusso di dati e selezionare un profilo del flusso di dati. Il profilo predefinito è selezionato per impostazione predefinita.
Costruisci la tua pipeline aggiungendo elementi alla tela.
1. Aggiungere un'origine: selezionare l'endpoint di origine e configurare gli argomenti per sottoscrivere i messaggi in arrivo.
2. Aggiungi trasformazioni: selezionare una o più trasformazioni per elaborare i dati. Le trasformazioni disponibili includono mappa, filtro, ramificatione, concatenazione, finestra e acceleratore. Per informazioni dettagliate su ogni tipo di trasformazione, vedere Panoramica dei grafici del flusso di dati.
3. Aggiungere una destinazione: selezionare l'endpoint di destinazione e configurare l'argomento o il percorso in cui inviare i dati elaborati.
Connettere gli elementi nell'ordine in cui si desidera che i dati vengano trasmessi.
Selezionare Salva per distribuire il grafico del flusso di dati.

La interfaccia della riga di comando di Azure utilizza un grafo di flusso dati proveniente da un singolo file di configurazione JSON che contiene tutti i nodi e le connessioni. Usare az iot ops dataflowgraph apply per creare o sostituire il grafico. Il seguente esempio legge i dati di temperatura, li converte in Fahrenheit e li invia a un argomento di destinazione.

Creare un graph.json file con le proprietà del grafico del flusso di dati. Nel graph.json file, memorizza le regole di ogni trasformazione nel value campo come una stringa JSON sfuggita. Per il formato leggibile delle regole di ogni trasformazione, vedere la procedura per tale tipo di trasformazione.

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

Applicare il file di configurazione.

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

Il grafico utilizza il profilo default di flusso dei dati. Per usare un profilo diverso, aggiungere --profile <PROFILE_NAME>.

Creare un file Bicep .bicep con la struttura seguente. In questo esempio viene creato un grafico del flusso di dati che legge i dati relativi alla temperatura, li converte in Fahrenheit e li invia a un argomento di destinazione.

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

Distribuire il file Bicep:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

Creare un file manifesto .yaml Kubernetes con la struttura seguente. In questo esempio viene creato un grafico del flusso di dati che legge i dati relativi alla temperatura, li converte in Fahrenheit e li invia a un argomento di destinazione.

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

Applicare il manifesto:

kubectl apply -f main.yaml

Configurare l'origine

L'origine definisce dove i dati entrano nella pipeline. Specificare un riferimento all'endpoint e uno o più argomenti.

Nell'editor del grafico del flusso di dati selezionare l'elemento di origine e configurare:

Impostazione	Descrizione
Punto finale	Endpoint del flusso di dati da usare. Selezionare il valore predefinito per il broker MQTT locale.
Topics	Uno o più argomenti a cui sottoscrivere i messaggi in arrivo.

La CLI applica l'intero grafo in una sola volta, quindi configura l'origine come nodo Source nel file di configurazione graph.json, quindi esegui 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'
    ]
  }
}

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

Aggiungere trasformazioni

Trasforma i dati del processo tra l'origine e la destinazione. Ogni trasformazione fa riferimento a un artefatto integrato e ha regole di configurazione.

Le trasformazioni predefinite disponibili sono:

Trasformazione	Artefatto	Descrizione
Mappa	`azureiotoperations/graph-dataflow-map:1.0.0`	Rinominare, ristrutturare, calcolare e copiare i campi.
Filter	`azureiotoperations/graph-dataflow-filter:1.0.0`	Eliminare i messaggi che corrispondono a una condizione.
Branch	`azureiotoperations/graph-dataflow-branch:1.0.0`	Instrada ogni messaggio verso un percorso `true` o `false` in base a una condizione.
Concatenate	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	Unire nuovamente due o più percorsi in un unico percorso.
Window	`azureiotoperations/graph-dataflow-window:1.0.0`	Raccogliere i messaggi in un intervallo di tempo, quindi aggregare.
Limitare	`azureiotoperations/graph-dataflow-throttle:1.0.0`	Limita la velocità dei messaggi per ogni pattern di argomento MQTT.

Per altre informazioni sull'arricchimento dei messaggi con dati esterni, vedere Arricchire i dati esterni.

Nell'editor del grafico del flusso di dati selezionare Aggiungi trasformazione e scegliere il tipo di trasformazione. Configurare le regole nell'editor visivo.

Ogni trasformazione è un nodo con nodeType impostato su Graph nel graph.json file di configurazione. Le regole della trasformazione sono un oggetto JSON, come questa mappa che converte la temperatura in Fahrenheit:

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

La configuration proprietà prende queste regole come stringa, quindi sfuggi al JSON delle regole e mettila nel value campo. Applicare il grafico completo con 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)\"}]}"
      }
    ]
  }
}

Suggerimento

Per generare la stringa sfuggita, salva le regole in un file come rules.json, esegui jq -c . rules.json, e incolla l'output a singola riga nel value campo.

Ogni trasformazione è un nodo con nodeType: 'Graph'. La configuration proprietà passa le regole come stringa JSON:

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

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

Ogni trasformazione è un nodo con nodeType: Graph. La configuration proprietà passa le regole come stringa JSON:

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

Concatenare più trasformazioni

Concatena qualsiasi numero di trasformazioni. Connetterli nella nodeConnections sezione nell'ordine in cui si vuole che i dati vengano trasmessi:

Trascinare le connessioni tra trasformazioni nell'area di disegno per definire l'ordine di elaborazione.

Definire l'ordine di elaborazione nella nodeConnections sezione del file di graph.json configurazione:

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

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

Configurare la destinazione

La destinazione definisce dove vanno i dati elaborati. Specificare un riferimento all'endpoint e un argomento o un percorso.

Selezionare l'elemento di destinazione e configurare:

Impostazione	Descrizione
Punto finale	Endpoint del flusso di dati a cui inviare i dati.
Topic	Argomento o percorso in cui pubblicare i dati elaborati.

Configura la destinazione come un nodo Destination nel file di configurazione graph.json, quindi applica il grafo completo con 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'
  }
}

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

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

Per il routing di argomenti dinamici in base al contenuto dei messaggi, vedere Instradare i messaggi a diversi argomenti.

Verificare che il grafico del flusso di dati funzioni

Dopo aver distribuito un grafo del flusso di dati, verificare che sia in esecuzione:

Nell'esperienza operativa, seleziona il tuo grafico di flusso dati per visualizzarne lo stato. Un grafico integro mostra uno stato In esecuzione.

Usare az iot ops dataflowgraph show per visualizzare i dettagli di un grafico:

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Per elencare tutti i grafici del flusso di dati associati a un profilo, usare az iot ops dataflowgraph list:

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Controllare lo stato della DataflowGraph risorsa:

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Importante

L'uso dei manifesti di distribuzione Kubernetes non è supportato negli ambienti di produzione e deve essere usato solo per il debug e il test.

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

Controllare i log dei pod per eventuali errori:

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

Commenti e suggerimenti

Questa pagina è stata utile?

Last updated on 2026-07-31

Creare un grafo del flusso di dati nelle operazioni IoT di Azure

Prerequisiti

Creare un grafo del flusso di dati

Configurare l'origine

Aggiungere trasformazioni

Concatenare più trasformazioni

Configurare la destinazione

Verificare che il grafico del flusso di dati funzioni

Contenuti correlati

Commenti e suggerimenti

Risorse aggiuntive