Creare flussi di dati nelle operazioni IoT di Azure

2025-06-22

Importante

Questa pagina include istruzioni sulla gestione dei componenti di Operazioni di Azure IoT tramite i manifesti di distribuzione Kubernetes, disponibile in anteprima. Questa funzionalità è fornita con diverse limitazioni e non deve essere usata per carichi di lavoro di produzione.

Vedere le condizioni per l'utilizzo supplementari per le anteprime di Microsoft Azure per termini legali aggiuntivi che si applicano a funzionalità di Azure in versione beta, in anteprima o in altro modo non ancora disponibili a livello generale.

Un flusso di dati è il percorso intrapreso dai dati dall'origine alla destinazione con trasformazioni facoltative. È possibile configurare il flusso di dati creando una risorsa personalizzata flusso di dati o usando l'interfaccia utente Web dell'esperienza operativa. Un flusso di dati è costituito da tre parti: l'origine, la trasformazione e la destinazione.

Diagramma di un flusso di dati che mostra il flusso dall'origine alla trasformazione e quindi alla destinazione.

Per definire l'origine e la destinazione, è necessario configurare gli endpoint del flusso di dati. La trasformazione è facoltativa e può includere operazioni come l'arricchimento dei dati, il filtro dei dati e il mapping dei dati a un altro campo.

Importante

Ogni flusso di dati deve avere l'endpoint predefinito del broker MQTT locale di Operazioni di Azure IoT come origine oppure come destinazione.

È possibile usare l'esperienza operativa in Operazioni di Azure IoT per creare un flusso di dati. L'esperienza operativa fornisce un'interfaccia visiva per configurare il flusso di dati. È anche possibile usare Bicep per creare un flusso di dati usando un file Bicep o usare Kubernetes per creare un flusso di dati usando un file YAML.

Continuare a leggere per informazioni su come configurare l'origine, la trasformazione e la destinazione.

Prerequisiti

È possibile distribuire i flussi di dati non appena si dispone di un'istanza di Operazioni di Azure IoT usando il profilo e l'endpoint del flusso di dati predefiniti. Tuttavia, è possibile configurare profili ed endpoint del flusso di dati per personalizzare il flusso di dati.

Profilo del flusso di dati

Se non si necessita di impostazioni di ridimensionamento diverse per i flussi di dati, usare il profilo del flusso di dati predefinito fornito da Operazioni di Azure IoT. Per informazioni su come configurare un nuovo profilo del flusso di dati, vedere Configurare i profili del flusso di dati.

Endpoint del flusso di dati

Gli endpoint del flusso di dati sono necessari per configurare l'origine e la destinazione per il flusso di dati. Per iniziare rapidamente, è possibile usare l'endpoint del flusso di dati predefinito per il broker MQTT locale. È anche possibile creare altri tipi di endpoint del flusso di dati, ad esempio Kafka, Hub eventi o Azure Data Lake Storage. Per informazioni su come configurare ogni tipo di endpoint del flusso di dati, vedere Configurare gli endpoint del flusso di dati.

Introduzione

Dopo aver ottenuto i prerequisiti, è possibile iniziare a creare un flusso di dati.

Per creare un flusso di dati nell'esperienza operativa, selezionare Flusso di dati>Crea flusso di dati.
Selezionare il nome segnaposto new-data-flow per impostare le proprietà del flusso di dati. Immettere il nome del flusso di dati e scegliere il profilo del flusso di dati da usare. Il profilo del flusso di dati predefinito è selezionato per impostazione predefinita. Per altre informazioni sui profili del flusso di dati, vedere Configurare il profilo del flusso di dati.

Importante

È possibile scegliere il profilo del flusso di dati solo durante la creazione di un flusso di dati. Non è possibile modificare il profilo del flusso di dati dopo la creazione del flusso di dati. Se si vuole modificare il profilo del flusso di dati di un flusso di dati esistente, eliminare il flusso di dati originale e crearne uno nuovo con il nuovo profilo del flusso di dati.
Configurare l'endpoint di origine, trasformazione e destinazione per il flusso di dati selezionando gli elementi nel diagramma del flusso di dati.

Usare il comando az iot ops dataflow apply per creare o modificare un flusso di dati.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

Il --config-file parametro è il percorso e il nome file di un file di configurazione JSON contenente le proprietà della risorsa.

In questo esempio si supponga che un file di configurazione denominato data-flow.json con il contenuto seguente archiviato nella home directory dell'utente:

{
    "mode": "Enabled",
    "operations": [
        {
            "operationType": "Source",
            "sourceSettings": {
                // See source configuration section
            }
        },
        {
            "operationType": "BuiltInTransformation",
            "builtInTransformationSettings": {
                // See transformation configuration section
            }
        },
        {
            "operationType": "Destination",
            "destinationSettings": {
                // See destination configuration section
            }
        }
    ]
}

Ecco un comando di esempio per creare o aggiornare un flusso di dati usando il profilo del flusso di dati predefinito:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

Creare un file Bicep .bicep per iniziare a creare un flusso di dati. Questo esempio mostra la struttura del flusso di dati contenente le configurazioni di origine, trasformazione e destinazione.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // See source configuration section
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // See transformation configuration section
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // See destination configuration section
        }
      }
    ]
  }
}

Creare un file manifesto Kubernetes .yaml per iniziare a creare un flusso di dati. Questo esempio mostra la struttura del flusso di dati contenente le configurazioni di origine, trasformazione e destinazione.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  # This field is required when configuring via Kubernetes YAML
  # The syntax is different when using Bicep
  profileRef: default 
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
      # Transformation optional
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

Vedere le sezioni seguenti per informazioni su come configurare i tipi di operazione del flusso di dati.

Origine

Per configurare un'origine per il flusso di dati, specificare il riferimento all'endpoint e un elenco di origini dati per l'endpoint. Scegliere una delle opzioni seguenti come origine per il flusso di dati.

Se l'endpoint predefinito non viene usato come origine, deve necessariamente essere usato come destinazione. Per altre informazioni sull'uso dell'endpoint broker MQTT locale, vedere Flussi di dati che devono usare l'endpoint broker MQTT locale.

Opzione 1: usare l'endpoint predefinito di Message Broker come origine

In Dettagli origine, selezionare Broker messaggi.

Immettere le impostazioni seguenti per l'origine del broker di messaggi:

Impostazione	Descrizione
Endpoint del flusso di dati	Selezionare l'impostazione predefinita per usare l'endpoint predefinito del broker messaggi MQTT.
Argomento	Filtro dell'argomento a cui sottoscrivere per i messaggi in arrivo. Usare Argomento/i>Aggiungi riga per aggiungere più argomenti. Per altre informazioni sugli argomenti, vedere Configurare argomenti MQTT o Kafka.
Schema del messaggio	Schema da utilizzare per deserializzare i messaggi in arrivo. Vedere Specificare lo schema per deserializzare i dati.

Selezionare Applica.

Ecco un esempio di configurazione dell'endpoint di origine per l'endpoint broker MQTT predefinito:

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "thermostats/+/sensor/temperature/#",
      "humidifiers/+/sensor/humidity/#"
    ],
    "endpointRef": "default"
  }
}

L'endpoint del broker di messaggi è configurato nel file Bicep. Ad esempio, l'endpoint seguente è un'origine per il flusso di dati.

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/sensor/temperature/#'
    'humidifiers/+/sensor/humidity/#'
  ]
}

Ad esempio, per configurare un'origine usando un endpoint di broker messaggi e due filtri di argomento, usare la configurazione seguente:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/sensor/temperature/#
    - humidifiers/+/sensor/humidity/#

Poiché dataSources consente di specificare argomenti MQTT o Kafka senza modificare la configurazione dell'endpoint, è possibile riutilizzare l'endpoint per più flussi di dati anche se gli argomenti sono diversi. Per altre informazioni, vedere Configurare origini dati.

Opzione 2: Usare asset come origine

È possibile usare un asset come origine per il flusso di dati. L'uso di un asset come origine è disponibile solo nell'esperienza operativa.

In Dettagli origine, selezionare Asset.
Selezionare l'asset da usare come endpoint di origine.
Selezionare Continua.

Verrà visualizzato un elenco di punti dati per l'asset selezionato.
Selezionare Applica per usare l'asset come endpoint di origine.

Quando si usa un asset come origine, la definizione dell'asset viene usata per dedurre lo schema per il flusso di dati. La definizione dell'asset include lo schema per i punti dati dell'asset. Per altre informazioni, vedere Gestire le configurazioni degli asset in modalità remota.

Una volta configurati, i dati dell'asset raggiungono il flusso di dati tramite il broker MQTT locale. Quindi, quando si usa una risorsa come sorgente, il flusso di dati usa in realtà l'endpoint predefinito del broker MQTT locale come sorgente.

Opzione 3: usare l'endpoint del flusso di dati MQTT o Kafka personalizzato come origine

Se è stato creato un endpoint flusso di dati MQTT o Kafka personalizzato (ad esempio, da usare con Griglia di eventi o Hub eventi), è possibile usarlo come origine per il flusso di dati. Tenere presente che gli endpoint di tipi di archiviazione, ad esempio Data Lake o Fabric OneLake, non possono essere usati come origine.

In Dettagli origine, selezionare Broker messaggi.

Immettere le impostazioni seguenti per l'origine del broker di messaggi:

Impostazione	Descrizione
Endpoint del flusso di dati	Usare il pulsante Riseleziona per selezionare un endpoint del flusso di dati MQTT o Kafka personalizzato. Per altre informazioni, vedere Configurare endpoint del flusso di dati MQTT o Configurare endpoint del flusso di dati di Hub eventi di Azure e Kafka.
Argomento	Filtro dell'argomento a cui sottoscrivere per i messaggi in arrivo. Usare Argomento/i>Aggiungi riga per aggiungere più argomenti. Per altre informazioni sugli argomenti, vedere Configurare argomenti MQTT o Kafka.
Schema del messaggio	Schema da utilizzare per deserializzare i messaggi in arrivo. Vedere Specificare lo schema per deserializzare i dati.

Selezionare Applica.

Sostituire i valori segnaposto con il nome e gli argomenti dell'endpoint personalizzati.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_1>",
            "<TOPIC_2>"
        ]
    }
}

Sostituire i valori segnaposto con il nome e gli argomenti dell'endpoint personalizzati.

sourceSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_1>'
    '<TOPIC_2>'
    // See section on configuring MQTT or Kafka topics for more information
  ]
}

Sostituire i valori segnaposto con il nome e gli argomenti dell'endpoint personalizzati.

sourceSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataSources:
  - <TOPIC_1>
  - <TOPIC_2>
  # See section on configuring MQTT or Kafka topics for more information

Configurare le origini dati (argomenti MQTT o Kafka)

È possibile specificare più argomenti MQTT o Kafka in un'origine senza dover modificare la configurazione dell'endpoint del flusso di dati. Questa flessibilità significa che lo stesso endpoint può essere riutilizzato in più flussi di dati, anche se gli argomenti variano. Per altre informazioni, vedere Riutilizzare gli endpoint del flusso di dati.

Argomenti relativi a MQTT

Quando l'origine è un endpoint MQTT (Griglia di eventi inclusa), è possibile usare il filtro dell'argomento MQTT per sottoscrivere a messaggi in arrivo. Il filtro dell'argomento può includere caratteri jolly per sottoscrivere a più argomenti. Ad esempio, thermostats/+/sensor/temperature/# sottoscrive tutti i messaggi dei sensori di temperatura dai termostati. Per configurare i filtri degli argomenti MQTT:

Nei dettagli di origine del flusso di dati dell'esperienza operativa, selezionare Broker messaggi, quindi usare il campo Argomento/i per specificare i filtri degli argomenti MQTT da sottoscrivere per i messaggi in arrivo. È possibile aggiungere più argomenti MQTT selezionando Aggiungi riga e immettendo un nuovo argomento.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<MESSAGE_BROKER_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_FILTER_1>",
            "<TOPIC_FILTER_2>"
            // Add more topic filters as needed
        ]
    }
}

Esempio con più filtri di argomento MQTT con caratteri jolly:

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
        ]
    }
}

In questo caso, il carattere jolly + viene usato per selezionare tutti i dispositivi negli argomenti thermostats e humidifiers. Il carattere jolly # viene usato per selezionare tutti i messaggi del sensore in tutti gli argomenti secondari temperature e humidity.

sourceSettings: {
  endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_FILTER_1>'
    '<TOPIC_FILTER_2>'
    // Add more topic filters as needed
  ]
}

Esempio con più filtri di argomento MQTT con caratteri jolly:

sourceSettings: {
  endpointRef: 'default'
  dataSources: [
    'thermostats/+/sensor/temperature/#'
    'humidifiers/+/sensor/humidity/#'
  ]
}

sourceSettings:
  endpointRef: <ENDPOINT_NAME>
  dataSources:
    - <TOPIC_FILTER_1>
    - <TOPIC_FILTER_2>
    # Add more topic filters as needed

Esempio con più filtri di argomento con caratteri jolly:

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/sensor/temperature/#
    - humidifiers/+/sensor/humidity/#

In questo caso, il carattere jolly + viene usato per selezionare tutti i dispositivi negli argomenti thermostats e humidifiers. Il carattere jolly # viene usato per selezionare tutti i messaggi in tutti gli argomenti secondari degli argomenti temperature e humidity.

Sottoscrizioni condivise

Per usare sottoscrizioni condivise con origini di broker messaggi, è possibile specificare l'argomento di sottoscrizione condivisa sotto forma di $shared/<GROUP_NAME>/<TOPIC_FILTER>.

In Dettagli di origine del flusso di dati dell'esperienza operativa, selezionare Broker messaggi e usare il campo Argomento per specificare il gruppo di sottoscrizioni e l'argomento condivisi.

{
  "operationType": "Source",
  "sourceSettings": {
    "dataSources": [
      "$shared/<GROUP_NAME>/<TOPIC_FILTER>"
    ]
  }
}

sourceSettings: {
  dataSources: [
    '$shared/<GROUP_NAME>/<TOPIC_FILTER>'
  ]
}

sourceSettings:
  dataSources:
    - $shared/<GROUP_NAME>/<TOPIC_FILTER>

Se il numero di istanze nel profilo del flusso di dati è maggiore di uno, la sottoscrizione condivisa viene abilitata automaticamente per tutti i flussi di dati che usano un'origine broker messaggi. In questo caso, viene aggiunto il prefisso $shared e il nome del gruppo di sottoscrizioni condiviso viene generato automaticamente. Ad esempio, se si dispone di un profilo flusso di dati con un numero di istanze pari a 3 e il flusso di dati usa un endpoint di broker messaggi come origine configurata con gli argomenti topic1 e topic2, questi vengono convertiti automaticamente in sottoscrizioni condivise come $shared/<GENERATED_GROUP_NAME>/topic1 e $shared/<GENERATED_GROUP_NAME>/topic2.

È possibile creare in modo esplicito un argomento denominato $shared/mygroup/topic nella configurazione. Tuttavia, l'aggiunta esplicita dell'argomento $shared non è consigliata perché il prefisso $shared viene aggiunto automaticamente quando necessario. I flussi di dati possono eseguire ottimizzazioni con il nome del gruppo, se non impostato. Ad esempio, $share non è impostato e i flussi di dati possono essere eseguiti solo sul nome dell'argomento.

Importante

I flussi di dati che richiedono una sottoscrizione condivisa quando il numero di istanze è superiore a uno sono importanti quando si utilizza il broker MQTT Griglia di eventi come sorgente, poiché non supporta le sottoscrizioni condivise. Per evitare la perdita di messaggi, impostare il numero di istanze del profilo del flusso di dati a uno quando si usa il broker MQTT di Griglia di eventi come sorgente. Vale a dire quando il flusso di dati è il sottoscrittore e riceve messaggi dal cloud.

Argomenti di Kafka

Quando l'origine è un endpoint Kafka (Hub eventi incluso), specificare i singoli argomenti Kafka a cui sottoscrivere per i messaggi in arrivo. I caratteri jolly non sono supportati, pertanto è necessario specificare ogni argomento in modo statico.

Annotazioni

Quando si usano Hub eventi tramite l'endpoint Kafka, ogni singolo hub eventi all'interno dello spazio dei nomi è l'argomento Kafka. Ad esempio, se si dispone di uno spazio dei nomi di Hub eventi con due hub eventi thermostats e humidifiers, è possibile specificare ogni hub eventi come argomento Kafka.

Per configurare gli argomenti Kafka:

Nei dettagli di origine del flusso di dati dell'esperienza operativa, selezionare Broker messaggi, quindi usare il campo Argomenti per specificare i filtri degli argomenti Kafka da sottoscrivere per i messaggi in arrivo.

Annotazioni

Nell'esperienza operativa, è possibile specificare un solo filtro di argomenti. Per usare più filtri di argomento, usare Bicep o Kubernetes.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "<KAFKA_ENDPOINT_NAME>",
    "dataSources": [
      "<KAFKA_TOPIC_1>",
      "<KAFKA_TOPIC_2>"
      // Add more Kafka topics as needed
    ]
  }
}

sourceSettings: {
  endpointRef: '<KAFKA_ENDPOINT_NAME>'
  dataSources: [
    '<KAFKA_TOPIC_1>'
    '<KAFKA_TOPIC_2>'
    // Add more Kafka topics as needed
  ]
}

sourceSettings:
  endpointRef: <KAFKA_ENDPOINT_NAME>
  dataSources:
    - <KAFKA_TOPIC_1>
    - <KAFKA_TOPIC_2>
    # Add more Kafka topics as needed

Specificare lo schema di origine

Quando si usa MQTT o Kafka come origine, è possibile specificare uno schema per visualizzare l'elenco di punti dati nell'interfaccia utente Web dell'esperienza operativa. L'uso di uno schema per deserializzare e convalidare i messaggi in ingresso non è attualmente supportato.

Se l'origine è un asset, lo schema viene dedotto automaticamente dalla definizione dell'asset.

Suggerimento

Per generare lo schema da un file di dati di esempio, usare l'helper Gen dello schema.

Per configurare lo schema usato per deserializzare i messaggi in ingresso da un'origine:

In Dettagli di origine del flusso di dati dell'esperienza operativa, selezionare Broker messaggi e usare il campo Schema messaggio per specificare lo schema. È possibile usare prima il pulsante Carica per caricare un file di schema. Per altre informazioni, vedere Familiarizzarsi con gli schemi di messaggi.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "<ENDPOINT_NAME>",
    "serializationFormat": "Json",
    "schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>"
  }
}

Dopo aver usato il registro schemi per archiviare lo schema, è possibile farvi riferimento nella configurazione del flusso di dati.

sourceSettings: {
  serializationFormat: 'Json'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}

Dopo aver usato il registro schemi per archiviare lo schema, è possibile farvi riferimento nella configurazione del flusso di dati.

sourceSettings:
  serializationFormat: Json
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'

Per altre informazioni, vedere Familiarizzarsi con gli schemi di messaggi.

Trasformazione

L'operazione di trasformazione consente di trasformare i dati dall'origine prima di inviarli alla destinazione. Le trasformazioni sono facoltative. Se non si necessita di apportare modifiche ai dati, non includere l'operazione di trasformazione nella configurazione del flusso di dati. Più trasformazioni vengono concatenate in fasi indipendentemente dall'ordine in cui sono specificate nella configurazione. L'ordine delle fasi è sempre:

Arricchimento: aggiungere ulteriori dati ai dati di origine in base a un set di dati e a una condizione per la corrispondenza.
Filtro: filtra i dati in base a una condizione.
Mappa, Calcola, Rinomina, o aggiungi una Nuova proprietà: sposta dati da un campo ad un altro con una conversione opzionale.

Questa sezione presenta un'introduzione alle trasformazioni del flusso di dati. Per informazioni più dettagliate, vedere Eseguire il mapping di dati usando flussi di dati, Convertire dati usando conversioni dei flussi di dati e Arricchire dati tramite flussi di dati.

Nell'esperienza operativa, selezionare Flusso di dati>Aggiungi trasformazione (facoltativo).

Screenshot dell'interfaccia dell'esperienza operativa che mostra l'aggiunta di una fase di trasformazione a un flusso di dati.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "datasets": [
      // See section on enriching data
    ],
    "filter": [
      // See section on filtering data
    ],
    "map": [
      // See section on mapping data
    ]
  }
}

builtInTransformationSettings: {
  datasets: [
    // See section on enriching data
  ]
  filter: [
    // See section on filtering data
  ]
  map: [
    // See section on mapping data
  ]
}

builtInTransformationSettings:
  datasets:
    # See section on enriching data
  filter:
    # See section on filtering data
  map:
    # See section on mapping data

Arricchire: Aggiungere dati di riferimento

Per arricchire i dati, aggiungere prima di tutto il set di dati di riferimento nell'archivio stati di Operazioni di Azure IoT. Il set di dati viene usato per aggiungere dati aggiuntivi ai dati di origine in base a una condizione. La condizione viene specificata come campo nei dati di origine che corrispondono a un campo nel set di dati.

È possibile caricare dati di esempio nell'archivio stati usando l'interfaccia della riga di comando dell'archivio stati. I nomi delle chiavi nell'archivio stati corrispondono a un set di dati nella configurazione del flusso di dati.

Attualmente, la fase Enrich non è supportata nell'esperienza operativa.

Per arricchire i dati, è possibile usare la builtInTransformationSettings proprietà nella configurazione del flusso di dati. La datasets proprietà viene utilizzata per specificare i set di dati da usare per l'arricchimento.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "datasets": [
      {
        "key": "<DATASET_KEY>",
        "inputs": [
          "$source.<SOURCE_FIELD>" // ---------------- $1
          "$context(<DATASET_KEY>).<DATASET_FIELD>" // - $2
        ],
        "expression": "$1 == $2"
      }
    ]
  }
}

Questo esempio illustra come usare il campo deviceId nei dati di origine per trovare una corrispondenza con il campo asset nel set di dati:

builtInTransformationSettings: {
  datasets: [
    {
      key: 'assetDataset'
      inputs: [
        '$source.deviceId' // ---------------- $1
        '$context(assetDataset).asset' // ---- $2
      ]
      expression: '$1 == $2'
    }
  ]
}

Ad esempio, è possibile usare il campo deviceId nei dati di origine per trovare la corrispondenza con il campo asset nel set di dati:

builtInTransformationSettings:
  datasets:
  - key: assetDataset
    inputs:
      - $source.deviceId # ------------- $1
      - $context(assetDataset).asset # - $2
    expression: $1 == $2

Se il set di dati ha un record con il campo asset, simile al seguente:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

I dati dell'origine con il campo deviceId corrispondente a thermostat1 hanno i campi location e manufacturer disponibili nelle fasi di filtro e mappa.

Per altre informazioni sulla sintassi delle condizioni, vedere Arricchire dati usando flussi di dati e Convertire dati tramite flussi di dati.

Filtro: filtrare i dati in base a una condizione

Per filtrare i dati in base a una condizione, è possibile usare la fase filter. La condizione viene specificata come campo nei dati di origine che corrispondono a un valore.

In Trasforma (facoltativo), selezionare Filtro>Aggiungi.
Immettere le impostazioni necessarie.

Impostazione Descrizione

Condizione di filtro Condizione per filtrare i dati in base a un campo nei dati di origine.

Descrizione Specificare una descrizione per la condizione di filtro.

Nel campo condizione di filtro, immettere @ o selezionare CTRL+ SPAZIO per scegliere i punti dati da un elenco a discesa.

È possibile immettere le proprietà dei metadati MQTT usando i formati @$metadata.user_properties.<property> o @$metadata.topic. È anche possibile immettere intestazioni $metadata usando il formato @$metadata.<header>. La sintassi $metadata è necessaria solo per le proprietà MQTT che fanno parte dell'intestazione del messaggio. Per altre informazioni, vedere riferimenti ai campi.

La condizione può utilizzare i campi nei dati di origine. Ad esempio, è possibile usare una condizione di filtro quale @temperature > 20 per filtrare i dati minori o uguali a 20 in base al campo della temperatura.
Selezionare Applica.

Impostazione	Descrizione
Condizione di filtro	Condizione per filtrare i dati in base a un campo nei dati di origine.
Descrizione	Specificare una descrizione per la condizione di filtro.

Ad esempio, è possibile usare il campo temperature nei dati di origine per filtrare i dati:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "filter": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "expression": "$1 > 20"
      }
    ]
  }
}

Ad esempio, è possibile usare il campo temperature nei dati di origine per filtrare i dati:

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      expression: '$1 > 20'
    }
  ]
}

Se il campo temperature è maggiore di 20, i dati vengono passati alla fase successiva. Se il campo temperature è minore o uguale a 20, i dati vengono filtrati.

Ad esempio, è possibile usare il campo temperature nei dati di origine per filtrare i dati:

builtInTransformationSettings:
  filter:
    - inputs:
      - temperature ? $last # - $1
      expression: "$1 > 20"

Se il campo temperature è maggiore di 20, i dati vengono passati alla fase successiva. Se il campo temperature è minore o uguale a 20, i dati vengono filtrati.

Mappa: spostare i dati da un campo a un altro

Per eseguire il mapping dei dati a un altro campo con la conversione facoltativa, è possibile usare l'operazione map. La conversione viene specificata come formula che utilizza i campi nei dati di origine.

Nell'esperienza operativa, il mapping è attualmente supportato tramite trasformazioniCalcola, Rinomina e Nuova proprietà.

Calcolo

È possibile usare la trasformazione Calcola per applicare una formula ai dati di origine. Questa operazione viene utilizzata per applicare una formula ai dati di origine e archiviare il campo dei risultati.

In Trasforma (facoltativo), selezionare Calcola>Aggiungi.

Immettere le impostazioni necessarie.

Impostazione	Descrizione
Selezionare la formula	Scegliere una formula esistente dall'elenco a discesa o selezionare Personalizzato per immetterne una manualmente.
Risultato	Specificare il nome visualizzato dell'output per il risultato.
Formula	Immettere la formula da applicare ai dati di origine.
Descrizione	Specificare una descrizione per la trasformazione.
Ultimo valore noto	Facoltativamente, è possibile usare l'ultimo valore noto nel caso in cui il valore corrente non fosse disponibile.

È possibile immettere o modificare una formula nel campo Formula. La formula può utilizzare i campi nei dati di origine. Immettere @ o selezionare CTRL+SPAZIO per scegliere i punti dati da un elenco a discesa. Per le formule predefinite, selezionare il segnaposto <dataflow> per visualizzare l'elenco dei punti dati disponibili.

È possibile immettere le proprietà dei metadati MQTT usando i formati @$metadata.user_properties.<property> o @$metadata.topic. È anche possibile immettere intestazioni $metadata usando il formato @$metadata.<header>. La sintassi $metadata è necessaria solo per le proprietà MQTT che fanno parte dell'intestazione del messaggio. Per altre informazioni, vedere riferimenti ai campi.

La formula può utilizzare i campi nei dati di origine. Ad esempio, è possibile usare il campo temperature nei dati di origine per convertire la temperatura in Celsius e archiviarla nel campo di output temperatureCelsius.

Selezionare Applica.

Rinominare

È possibile rinominare un punto dati usando la trasformazione Rinomina . Questa operazione viene utilizzata per rinominare un punto dati nei dati di origine con un nuovo nome. Il nuovo nome può essere usato nelle fasi successive del flusso di dati.

In Trasforma (facoltativo), selezionare Rinomina>Aggiungi.

Immettere le impostazioni necessarie.

Impostazione	Descrizione
Punto Dati	Selezionare un punto dati dall'elenco a discesa o immettere un'intestazione $metadata.
Nuovo nome del punto dati	Immettere il nuovo nome per il punto dati.
Descrizione	Specificare una descrizione per la trasformazione.

Selezionare Applica.

Nuova proprietà

È possibile aggiungere una nuova proprietà ai dati di origine usando la trasformazione Nuova proprietà. Questa operazione viene utilizzata per aggiungere una nuova proprietà ai dati di origine. La nuova proprietà può essere usata nelle fasi successive del flusso di dati.

In Trasforma (facoltativo), selezionare Nuova proprietà>Aggiungi.

Immettere le impostazioni necessarie.

Impostazione	Descrizione
Chiave proprietà	Immettere la chiave per la nuova proprietà.
Valore proprietà	Immettere il valore per la nuova proprietà.
Descrizione	Specificare una descrizione per la nuova proprietà.

Selezionare Applica.

Ad esempio, è possibile usare il campo temperature nei dati di origine per convertire la temperatura in Celsius e archiviarla nel campo temperatureCelsius. È anche possibile arricchire i dati di origine con il campo location dal set di dati di contestualizzazione:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "output": "temperatureCelsius",
        "expression": "($1 - 32) * 5/9"
      },
      {
        "inputs": [
          "$context(assetDataset).location" // - $2
        ],
        "output": "location"
      }
    ]
  }
}

È possibile accedere alle proprietà dei metadati MQTT usando i formati $metadata.user_properties.<property> o $metadata.topic. È anche possibile immettere intestazioni $metadata usando il formato $metadata.<header>. Per altre informazioni, vedere riferimenti ai campi.

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        'temperature'
      ]
      output: 'temperatureCelsius'
      expression: '($1 - 32) * 5/9'
    }
    {
      inputs: [
        '$context(assetDataset).location'
      ]
      output: 'location'
    }
  ]
}

builtInTransformationSettings:
  map:
    - inputs:
      - temperature # - $1
      expression: "($1 - 32) * 5/9"
      output: temperatureCelsius
    - inputs:
      - $context(assetDataset).location  
      output: location

Per altre informazioni, vedere Eseguire il mapping di dati usando flussi di dati e Convertire dati usando i flussi di dati.

Rimuovi

Per impostazione predefinita, tutti i punti dati sono inclusi nello schema di output. È possibile rimuovere qualsiasi punto dati dalla destinazione usando la trasformazione Rimuovi .

In Trasforma (facoltativo) selezionare Rimuovi.
Selezionare il punto dati da rimuovere dallo schema di output.
Selezionare Applica.

Per rimuovere un punto dati dallo schema di output, è possibile usare la builtInTransformationSettings proprietà nella configurazione del flusso di dati. La map proprietà viene utilizzata per specificare i punti dati da rimuovere.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "*"
        ],
        "output": "*"
      },
      {
        "inputs": [
          "weight"
        ],
        "output": ""
      }
      {
          "inputs": [
          "weight.SourceTimestamp"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.Value"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Code"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Symbol"
          ],
          "output": ""
      }
    ]
  }
}

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        '*'
      ]
      output: '*'
    }
    {
      inputs: [
        'weight'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.SourceTimestamp'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.Value'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Code'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Symbol'
      ]
      output: ''
    }
  ]
}

builtInTransformationSettings:
  map:
    - type: PassThrough
      inputs:
        - "*"
      output: "*"
    - inputs:
        - weight
      output: ""
    - inputs:
        - weight.SourceTimestamp
      output: ""
    - inputs:
        - weight.Value
      output: ""
    - inputs:
        - weight.StatusCode
      output: ""
    - inputs:
        - weight.StatusCode.Code
      output: ""
    - inputs:
        - weight.StatusCode.Symbol
      output: ""

Per altre informazioni, vedere Eseguire il mapping di dati usando flussi di dati e Convertire dati usando i flussi di dati.

Serializzare i dati in base a uno schema

Se si desidera serializzare i dati prima di inviarli alla destinazione, è necessario specificare uno schema e un formato di serializzazione. Altrimenti, i dati verranno serializzati in JSON con i tipi dedotti. Endpoint di archiviazione come Microsoft Fabric o Azure Data Lake richiedono uno schema per garantire la coerenza dei dati. I formati di serializzazione supportati sono Parquet e Delta.

Suggerimento

Per generare lo schema da un file di dati di esempio, usare l'helper Gen dello schema.

Per l'esperienza operativa, specificare lo schema e il formato di serializzazione nei dettagli dell'endpoint del flusso di dati. Gli endpoint che supportano i formati di serializzazione sono Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Esplora dati di Azure e archiviazione locale. Ad esempio, per serializzare i dati in formato Delta, è necessario caricare uno schema nel registro schemi e farvi riferimento nella configurazione dell'endpoint di destinazione del flusso di dati.

Screenshot che usa l'esperienza operativa per impostare la serializzazione dell'endpoint di destinazione del flusso di dati.

Dopo aver caricato uno schema nel registro schemi, è possibile farvi riferimento nella configurazione del flusso di dati.

{
  "builtInTransformationSettings": {
    "serializationFormat": "Delta",
    "schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>"
  }
}

Dopo aver caricato uno schema nel registro schemi, è possibile farvi riferimento nella configurazione del flusso di dati.

builtInTransformationSettings: {
  serializationFormat: 'Delta'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'
}

Dopo aver caricato uno schema nel registro schemi, è possibile farvi riferimento nella configurazione del flusso di dati.

builtInTransformationSettings:
  serializationFormat: Delta
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA>:<VERSION>'

Per altre informazioni sul registro schemi, vedere Informazioni sugli schemi di messaggi.

Destinazione

Per configurare una destinazione per il flusso di dati, specificare il riferimento all'endpoint e la destinazione dei dati. È possibile specificare un elenco di destinazioni dati per l'endpoint.

Per inviare dati a una destinazione diversa dal broker MQTT locale, creare un endpoint del flusso di dati. Per informazioni su come eseguire quest'operazione, vedere Configurare gli endpoint del flusso di dati. Se la destinazione non è il broker MQTT locale, deve essere necessariamente usata come origine. Per altre informazioni sull'uso dell'endpoint broker MQTT locale, vedere Flussi di dati che devono usare l'endpoint broker MQTT locale.

Importante

Gli endpoint di archiviazione richiedono uno schema per la serializzazione. Per usare il flusso di dati con Microsoft Fabric OneLake, Azure Data Lake Storage, Esplora dati di Azure o Archiviazione locale, è necessario specificare un riferimento allo schema.

Selezionare l'endpoint del flusso di dati da usare come destinazione.

Gli endpoint di archiviazione richiedono uno schema per la serializzazione. Se si sceglie un endpoint di destinazione di Microsoft Fabric OneLake, Azure Data Lake Storage, Esplora dati di Azure o Archiviazione locale, è necessario specificare un riferimento allo schema. Ad esempio, per serializzare i dati in un endpoint di Microsoft Fabric in formato Delta, è necessario caricare uno schema nel registro schemi e farvi riferimento nella configurazione dell'endpoint di destinazione del flusso di dati.
Selezionare Continua per configurare la destinazione.
Immettere le impostazioni necessarie per la destinazione, incluso l'argomento o la tabella a cui inviare i dati. Per altre informazioni, vedere Configurare la destinazione dati (argomento, contenitore o tabella).

{
  "destinationSettings": {
    "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
    "dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
  }
}

destinationSettings: {
  endpointRef: '<CUSTOM_ENDPOINT_NAME>'
  dataDestination: '<TOPIC_OR_TABLE>' // See section on configuring data destination
}

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE> # See section on configuring data destination

Configurare la destinazione dati (argomento, contenitore o tabella)

Analogamente alle origini dati, la destinazione dei dati è un concetto usato per mantenere riutilizzabili gli endpoint del flusso di dati in più flussi di dati. Essenzialmente, rappresenta la sottodirectory nella configurazione dell'endpoint del flusso di dati. Ad esempio, se l'endpoint del flusso di dati è un endpoint di archiviazione, la destinazione dei dati è la tabella nell'account di archiviazione. Se l'endpoint del flusso di dati è un endpoint Kafka, la destinazione dei dati è l'argomento Kafka.

Tipo di endpoint	Significato della destinazione dei dati	Descrizione
MQTT (o Griglia di eventi)	Argomento	Argomento MQTT a cui vengono inviati i dati. Sono supportati solo gli argomenti statici, senza caratteri jolly.
Kafka (o Hub eventi)	Argomento	Argomento Kafka a cui vengono inviati i dati. Sono supportati solo gli argomenti statici, senza caratteri jolly. Se l'endpoint è uno spazio dei nomi di Hub eventi, la destinazione dei dati è il singolo hub eventi all'interno dello spazio dei nomi.
Azure Data Lake Storage	Contenitore	Contenitore nell'account di archiviazione. Non la tabella.
Microsoft Fabric OneLake	Tabella o cartella	Corrisponde al tipo di percorso configurato per l'endpoint.
Esplora dati di Azure	Tabella	Tabella nel database di Esplora dati di Azure.
Archiviazione locale	Cartella	Nome della cartella o della directory nel montaggio permanente del volume di archiviazione locale. Quando si usa Archiviazione di Container Azure abilitata dai volumi Edge di inserimento cloud di Azure Arc, questo deve corrispondere al parametro `spec.path` per il sottovolume creato.

Per configurare la destinazione dati:

Quando si usa l'esperienza operativa, il campo destinazione dati viene interpretato automaticamente in base al tipo di endpoint. Ad esempio, se l'endpoint del flusso di dati è un endpoint di archiviazione, la pagina dei dettagli di destinazione richiede di immettere il nome del contenitore. Se l'endpoint del flusso di dati è un endpoint MQTT, la pagina dei dettagli di destinazione richiede di immettere l'argomento, e così via.

Screenshot che mostra l'esperienza operativa che richiede all'utente di immettere un argomento MQTT in base al tipo di endpoint.

{
  "destinationSettings": {
    "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
    "dataDestination": "<TOPIC_OR_TABLE>" // See section on configuring data destination
  }
}

Ad esempio, per inviare dati al broker MQTT locale su un topic MQTT statico, usare la seguente configurazione:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "example-topic"
  }
}

In alternativa, se si dispone di un endpoint dell'hub eventi personalizzato, la configurazione sarà simile alla seguente:

{
  "destinationSettings": {
    "endpointRef": "my-eh-endpoint",
    "dataDestination": "individual-event-hub"
  }
}

La sintassi è la stessa per tutti gli endpoint del flusso di dati:

destinationSettings: {
  endpointRef: "<CUSTOM_ENDPOINT_NAME>"
  dataDestination: '<TOPIC_OR_TABLE>'
}

Ad esempio, per inviare dati al broker MQTT locale su un topic MQTT statico, usare la seguente configurazione:

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'example-topic'
}

In alternativa, se si dispone di un endpoint dell'hub eventi personalizzato, la configurazione sarà simile alla seguente:

destinationSettings: {
  endpointRef: 'my-eh-endpoint'
  dataDestination: 'individual-event-hub'
}

Un altro esempio usa un endpoint di archiviazione come destinazione:

destinationSettings: {
  endpointRef: 'my-adls-endpoint'
  dataDestination: 'my-container'
}

La sintassi è la stessa per tutti gli endpoint del flusso di dati:

destinationSettings:
  endpointRef: <CUSTOM_ENDPOINT_NAME>
  dataDestination: <TOPIC_OR_TABLE>

Ad esempio, per inviare dati al broker MQTT locale su un topic MQTT statico, usare la seguente configurazione:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

In alternativa, se si dispone di un endpoint dell'hub eventi personalizzato, la configurazione sarà simile alla seguente:

destinationSettings:
  endpointRef: my-eh-endpoint
  dataDestination: individual-event-hub

Un altro esempio usa un endpoint di archiviazione come destinazione:

destinationSettings:
  endpointRef: my-adls-endpoint
  dataDestination: my-container

Esempio

L'esempio seguente è una configurazione del flusso di dati che usa l'endpoint MQTT per l'origine e la destinazione. L'origine filtra i dati dall'argomento MQTT azure-iot-operations/data/thermostat. La trasformazione converte la temperatura in Fahrenheit e filtra i dati in cui la temperatura moltiplicata per l'umidità è minore di 100000. La destinazione invia i dati all'argomento MQTT factory.

Usare il comando az iot ops dataflow apply per creare o modificare un flusso di dati.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

Il --config-file parametro è il percorso e il nome file di un file di configurazione JSON contenente le proprietà della risorsa.

In questo esempio si supponga che un file di configurazione denominato data-flow.json con il contenuto seguente archiviato nella home directory dell'utente:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "builtInTransformationSettings": {
          "datasets": [],
          "filter": [
            {
              "expression": "$1 * $2 < 100000",
              "inputs": [
                "temperature.Value",
                "\"Tag 10\".Value"
              ],
              "type": "Filter"
            }
          ],
          "map": [
            {
              "inputs": [
                "*"
              ],
              "output": "*",
              "type": "PassThrough"
            },
            {
              "expression": "fToC($1)",
              "inputs": [
                "Temperature.Value"
              ],
              "output": "TemperatureF",
              "type": "Compute"
            },
            {
              "inputs": [
                "@\"Tag 10\".Value"
              ],
              "output": "Humidity",
              "type": "Rename"
            }
          ],
          "serializationFormat": "Json"
        },
        "operationType": "BuiltInTransformation"
      },
      {
        "destinationSettings": {
          "dataDestination": "factory",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Ecco un comando di esempio per creare o aggiornare un flusso di dati usando il profilo del flusso di dati predefinito:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param dataflowName string = '<DATAFLOW_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

// Pointer to the default data flow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default data flow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-11-01' = {
  // Reference to the parent data flow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // Use the default MQTT endpoint as the source
          endpointRef: defaultDataflowEndpoint.name
          // Filter the data from the MQTT topic azure-iot-operations/data/thermostat
          dataSources: [
            'azure-iot-operations/data/thermostat'
          ]
        }
      }
      // Transformation optional
      {
        operationType: 'BuiltInTransformation'
        builtInTransformationSettings: {
          // Filter the data where temperature * "Tag 10" < 100000
          filter: [
            {
              inputs: [
                'temperature.Value'
                '"Tag 10".Value'
              ]
              expression: '$1 * $2 < 100000'
            }
          ]
          map: [
            // Passthrough all values by default
            {
              inputs: [
                '*'
              ]
              output: '*'
            }
            // Convert temperature to Fahrenheit and output it to TemperatureF
            {
              inputs: [
                'temperature.Value'
              ]
              output: 'TemperatureF'
              expression: 'cToF($1)'
            }
          // Extract the "Tag 10" value and output it to Humidity
            {
              inputs: [
                '"Tag 10".Value'
              ]
              output: 'Humidity'
            }
          ]
        }
      }
      {
        operationType: 'Destination'
        destinationSettings: {
          // Use the default MQTT endpoint as the destination
          endpointRef: defaultDataflowEndpoint.name
          // Send the data to the MQTT topic factory
          dataDestination: 'factory'
        }
      }
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow profile
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # Use the default MQTT endpoint as the source
        endpointRef: default
        # Filter the data from the MQTT topic azure-iot-operations/data/thermostat
        dataSources:
          - azure-iot-operations/data/thermostat
    # Transformation optional
    - operationType: builtInTransformation
      builtInTransformationSettings:
        # Filter the data where temperature * "Tag 10" < 100000
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: '$1 * $2 < 100000'
        map:
          # Passthrough all values by default
          - inputs:
              - '*'
            output: '*'
          # Convert temperature to Fahrenheit and output it to TemperatureF
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          # Extract the "Tag 10" value and output it to Humidity
          - inputs:
              - '"Tag 10".Value'
            output: 'Humidity'
    - operationType: Destination
      destinationSettings:
        # Use the default MQTT endpoint as the destination
        endpointRef: default
        # Send the data to the MQTT topic factory
        dataDestination: factory

Per altri esempi di configurazioni del flusso di dati, vedere API REST di Azure - Flusso di dati e Avvio rapido su Bicep.

Verificare che un flusso di dati funzioni

Seguire l'esercitazione : Bridge MQTT bidirezionale a Griglia di eventi di Azure per verificare che il flusso di dati funzioni.

Esportare la configurazione del flusso di dati

Per esportare la configurazione del flusso di dati, è possibile usare l'esperienza operativa o esportando la risorsa personalizzata del flusso di dati.

Selezionare il flusso di dati da esportare, quindi selezionare Esporta sulla barra degli strumenti.

Screenshot dell'interfaccia dell'esperienza operativa che mostra l'opzione di esportazione per un flusso di dati configurato.

Usare il comando az iot ops dataflow show per esportare un flusso di dati.

az iot ops dataflow show --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <DataflowName> --profile <DataflowProfileName> --output json > my-dataflow.json

Ecco un comando di esempio per esportare un flusso di dati denominato data-flow in un file JSON denominato data-flow.json:

az iot ops dataflow show --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --output json > data-flow.json

kubectl get dataflow my-dataflow -o yaml > my-dataflow.yaml

Configurazione corretta del flusso di dati

Per assicurarsi che il flusso di dati funzioni come previsto, verificare quanto segue:

L'endpoint del flusso di dati MQTT predefinito deve essere usato come origine o destinazione.
Il profilo del flusso di dati è esistente e vi viene fatto riferimento nella configurazione del flusso di dati.
L'origine è un endpoint MQTT, un endpoint Kafka o un asset. Gli endpoint del tipo di archiviazione non possono essere usati come origine.
Quando si usa Griglia di eventi come origine, il numero di istanze del profilo del flusso di dati è impostato su 1, poiché il broker MQTT di Griglia di eventi non supporta le sottoscrizioni condivise.
Quando si usa Hub eventi come origine, ogni hub eventi nello spazio dei nomi è un argomento Kafka separato e deve essere specificato come origine dati.
La trasformazione, se usata, viene configurata con la sintassi corretta, incluso l'escape corretto di caratteri speciali.
Quando si usano endpoint di tipo di archiviazione come destinazione, viene specificato uno schema.

Condividi tramite

Creare flussi di dati nelle operazioni IoT di Azure

Prerequisiti

Profilo del flusso di dati

Endpoint del flusso di dati

Introduzione

Origine

Opzione 1: usare l'endpoint predefinito di Message Broker come origine

Opzione 2: Usare asset come origine

Opzione 3: usare l'endpoint del flusso di dati MQTT o Kafka personalizzato come origine

Configurare le origini dati (argomenti MQTT o Kafka)

Argomenti relativi a MQTT

Sottoscrizioni condivise

Argomenti di Kafka

Specificare lo schema di origine

Trasformazione

Arricchire: Aggiungere dati di riferimento

Filtro: filtrare i dati in base a una condizione

Mappa: spostare i dati da un campo a un altro

Calcolo

Rinominare

Nuova proprietà

Rimuovi

Serializzare i dati in base a uno schema

Destinazione

Configurare la destinazione dati (argomento, contenitore o tabella)

Esempio

Verificare che un flusso di dati funzioni

Esportare la configurazione del flusso di dati

Configurazione corretta del flusso di dati

Passaggi successivi

Commenti e suggerimenti

Risorse aggiuntive