Configurare i flussi di dati nelle operazioni di Azure IoT

Per creare un flusso di dati nell'esperienza operativa, selezionare Flusso>di dati Crea flusso di dati. Verrà quindi visualizzata la pagina in cui è possibile configurare l'origine, la trasformazione e la destinazione per il flusso di dati.

Creare un file Bicep .bicep per iniziare a creare un flusso di dati. In questo esempio viene illustrata 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-09-15-preview' existing = {
  name: aioInstanceName
}

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

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

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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 .yaml Kubernetes per iniziare a creare un flusso di dati. In questo esempio viene illustrata la struttura del flusso di dati contenente le configurazioni di origine, trasformazione e destinazione.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

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

1. In Dettagli origine selezionare Asset.

2. Selezionare l'asset da usare come endpoint di origine.

3. Selezionare Continua.

   Viene visualizzato un elenco di punti dati per l'asset selezionato.

   

4. Selezionare Applica per usare l'asset come endpoint di origine.

La configurazione di un asset come origine è disponibile solo nell'esperienza operativa.

La configurazione di un asset come origine è disponibile solo nell'esperienza operativa.

1. In Dettagli origine selezionare MQTT.

   

2. Immettere le impostazioni seguenti per l'origine MQTT:

   Impostazione	Descrizione
   Argomento MQTT	Filtro dell'argomento MQTT per sottoscrivere i messaggi in arrivo. Vedere Configurare MQTT o Kafka argomenti.
   Schema del messaggio	Schema da utilizzare per deserializzare i messaggi in arrivo. Vedere Specificare lo schema per deserializzare i dati.

3. Selezionare Applica.

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

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

In questo caso, dataSources è possibile specificare più argomenti MQTT o Kafka senza dover modificare la configurazione dell'endpoint. Questa flessibilità significa che lo stesso endpoint può essere riutilizzato in più flussi di dati, anche se gli argomenti variano. Per altre informazioni, vedere Configurare le origini dati.

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

sourceSettings:
  endpointRef: default
  dataSources:
    - thermostats/+/telemetry/temperature/#
    - humidifiers/+/telemetry/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 le origini dati.

L'uso di un endpoint MQTT o Kafka personalizzato come origine non è attualmente supportato nell'esperienza operativa.

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

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

Nei dettagli dell'origine del flusso di dati dell'esperienza operativa selezionare MQTT, quindi usare il campo dell'argomento MQTT per specificare il filtro dell'argomento MQTT da sottoscrivere per i messaggi in arrivo.

sourceSettings: {
  endpointRef: '<MQTT_ENDPOINT_NAME>'
  dataSources: [
    '<MQTT_TOPIC_FILTER_1>'
    '<MQTT_TOPIC_FILTER_2>'
    // Add more MQTT topic filters as needed
  ]
}

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

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

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

sourceSettings:
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources:
    - <MQTT_TOPIC_FILTER_1>
    - <MQTT_TOPIC_FILTER_2>
    # Add more MQTT topic filters as needed

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

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

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

In Dettagli origine flusso di dati dell'esperienza operativa selezionare MQTT e usare il campo argomento MQTT per specificare il gruppo di sottoscrizioni e l'argomento condivisi.

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

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

L'uso di un endpoint Kafka come origine non è attualmente supportato nell'esperienza operativa.

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

In Dettagli origine flusso di dati dell'esperienza operativa selezionare MQTT 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 Informazioni sugli schemi dei messaggi.

Dopo aver usato il Registro di sistema dello schema 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 di sistema dello schema per archiviare lo schema, è possibile farvi riferimento nella configurazione del flusso di dati.

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

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

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

Nell'esperienza operativa, la fase Enrich è attualmente supportata usando le trasformazioni di proprietà Rename e New.

1. Nell'esperienza operativa selezionare un flusso di dati e quindi Aggiungi trasformazione (facoltativo).

2. Scegliere Rinomina o Nuove trasformazioni di proprietà e quindi selezionare Aggiungi.

   

Questo esempio illustra come usare il deviceId campo nei dati di origine per trovare una corrispondenza con il asset campo 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

1. In Trasforma (facoltativo) selezionare Filtro>Aggiungi.

2. Scegliere i punti dati da includere nel set di dati.

3. Aggiungere una condizione di filtro e una descrizione.

   

4. Selezionare Applica.

Ad esempio, è possibile usare una condizione di filtro come temperature > 20 filtrare i dati minori o uguali a 20 in base al campo della temperatura.

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.

Nell'esperienza operativa il mapping è attualmente supportato tramite trasformazioni di calcolo .

1. In Trasforma (facoltativo) selezionare> Calcolo aggiungi.

2. Immettere i campi e le espressioni obbligatori.

   

3. 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:

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

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:

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

Attualmente, l'impostazione dello schema di output e della serializzazione non è supportata nell'esperienza operativa.

Dopo aver caricato uno schema nel Registro di sistema dello schema, è 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 di sistema dello schema, è possibile farvi riferimento nella configurazione del flusso di dati.

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

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

   

2. Selezionare Continua per configurare la destinazione.

3. 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

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.

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 i dati al broker MQTT locale un argomento MQTT statico, usare la configurazione seguente:

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 i dati al broker MQTT locale un argomento MQTT statico, usare la configurazione seguente:

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

Per l'esempio di configurazione, vedere schede Bicep o Kubernetes.

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

resource aioInstance 'Microsoft.IoTOperations/instances@2024-09-15-preview' existing = {
  name: aioInstanceName
}

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

// Pointer to the default dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-09-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-09-15-preview' = {
  // Reference to the parent dataflow 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/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow 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

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

Bicep è un'infrastruttura come codice e non è necessaria alcuna esportazione. Usare il file modello Bicep per creare un flusso di dati per configurare e configurare rapidamente i flussi di dati.

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