Configurare i flussi di dati nelle operazioni di Azure IoT

Importante

Anteprima di Operazioni di Azure IoT – Abilitata da Azure Arc è attualmente in anteprima. Non è consigliabile usare questo software di anteprima negli ambienti di produzione.

Sarà necessario distribuire una nuova installazione di Operazioni di Azure IoT quando sarà messa a disposizione una versione disponibile a livello generale. Non sarà possibile aggiornare un'installazione di anteprima.

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 che i dati prendono dall'origine alla destinazione con trasformazioni facoltative. È possibile configurare il flusso di dati creando una risorsa personalizzata flusso di dati o usando il portale di Azure IoT Operations Studio. Un flusso di dati è costituito da tre parti: l'origine, la trasformazione e la 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.

Questo articolo illustra come creare un flusso di dati con un esempio, tra cui l'origine, la trasformazione e la destinazione.

Prerequisiti

• Istanza di Azure IoT Operations Preview
• Profilo del flusso di dati configurato
• Endpoint del flusso di dati. Ad esempio, creare un endpoint del flusso di dati per il broker MQTT locale. È possibile usare questo endpoint sia per l'origine che per la destinazione. In alternativa, è possibile provare altri endpoint come 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.

Creare un flusso di dati

Dopo aver creato gli endpoint del flusso di dati, è possibile usarli per creare un flusso di dati. Tenere presente che un flusso di dati è costituito da tre parti: l'origine, la trasformazione e la destinazione.

Portale

Per creare un flusso di dati nel portale dell'esperienza operativa, selezionare Flusso>di dati Crea flusso di dati.

Kubernetes

La struttura complessiva di una configurazione del flusso di dati è la seguente:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

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

Configurare un'origine con un endpoint del flusso di dati per ottenere i dati

Per configurare un'origine per il flusso di dati, specificare il riferimento all'endpoint e l'origine dati. È possibile specificare un elenco di origini dati per l'endpoint.

Portale

Usare Asset come origine

È possibile usare un asset come origine per il flusso di dati. Questa opzione è disponibile solo nel portale dell'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.

Kubernetes

La configurazione di un asset come origine è disponibile solo nel portale dell'esperienza operativa.

Usare MQTT come origine

Portale

1. In Dettagli origine selezionare MQTT.

2. Immettere l'argomento MQTT a cui si vuole restare in ascolto per i messaggi in arrivo.

3. Scegliere uno schema messaggio dall'elenco a discesa o caricare un nuovo schema. Se i dati di origine hanno campi o campi facoltativi con tipi diversi, specificare uno schema di deserializzazione per garantire la coerenza. Ad esempio, i dati potrebbero avere campi che non sono presenti in tutti i messaggi. Senza lo schema, la trasformazione non può gestire questi campi perché avrebbero valori vuoti. Con lo schema è possibile specificare i valori predefiniti o ignorare i campi.

   

4. Selezionare Applica.

Kubernetes

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

sourceSettings:
  endpointRef: mq
  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 Riutilizzare gli endpoint del flusso di dati.

Specificare lo schema per deserializzare i dati

Se i dati di origine hanno campi o campi facoltativi con tipi diversi, specificare uno schema di deserializzazione per garantire la coerenza. Ad esempio, i dati potrebbero avere campi che non sono presenti in tutti i messaggi. Senza lo schema, la trasformazione non può gestire questi campi perché avrebbero valori vuoti. Con lo schema è possibile specificare i valori predefiniti o ignorare i campi.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Per specificare lo schema, creare il file e archiviarlo nel Registro di sistema dello schema.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Nota

L'unico formato di serializzazione supportato è JSON. Lo schema è facoltativo.

Per altre informazioni sul Registro di sistema dello schema, vedere Informazioni sugli schemi dei messaggi.

Sottoscrizioni condivise

Per usare le sottoscrizioni condivise con le origini MQTT, è possibile specificare l'argomento della sottoscrizione condivisa sotto forma di $shared/<subscription-group>/<topic>.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Nota

Se il numero di istanze nel profilo del flusso di dati è maggiore di 1, è necessario usare l'argomento della sottoscrizione condivisa.

Configurare la trasformazione per elaborare i dati

L'operazione di trasformazione consente di trasformare i dati dall'origine prima di inviarli alla destinazione. Le trasformazioni sono facoltative. Se non è necessario 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

1. Arricchimento: aggiungere dati aggiuntivi ai dati di origine in base a un set di dati e a una condizione per la corrispondenza.
2. Filtro: filtra i dati in base a una condizione.
3. Mappa: spostare i dati da un campo a un altro con una conversione facoltativa.

Portale

Nel portale dell'esperienza operativa selezionare Aggiungi trasformazione flusso>di dati (facoltativo) .

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

Arricchire: Aggiungere dati di riferimento

Per arricchire i dati, è possibile usare il set di dati di riferimento nell'archivio stati distribuito di Azure IoT Operations (DSS). 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.

I nomi delle chiavi nell'archivio di stati distribuiti corrispondono a un set di dati nella configurazione del flusso di dati.

Portale

Attualmente, l'operazione di arricchimento non è disponibile nel portale dell'esperienza operativa.

Kubernetes

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 la corrispondenza dei deviceId campi hanno i location campi e manufacturer disponibili in filter e map le thermostat1 fasi.

È possibile caricare dati di esempio nel DSS usando l'esempio di strumento set di DSS.

Per altre informazioni sulla sintassi delle condizioni, vedere Arricchire i dati usando flussi di dati e Convertire i 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.

Portale

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.

Kubernetes

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.

Portale

Nel portale dell'esperienza operativa il mapping è attualmente supportato usando le trasformazioni di calcolo .

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

2. Immettere i campi e le espressioni obbligatori.

   

3. Selezionare Applica.

Kubernetes

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
      output: temperatureCelsius
      expression: "($1 - 32) * 5/9"
    - inputs:
      - $context(assetDataset).location  
      output: location

Per altre informazioni, vedere Eseguire il mapping dei dati usando flussi di dati e Convertire i 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. In caso contrario, i dati vengono serializzati in JSON con i tipi dedotti. Tenere presente che gli endpoint di archiviazione come Microsoft Fabric o Azure Data Lake richiedono uno schema per garantire la coerenza dei dati.

Portale

Specificare lo schema di output quando si aggiunge l'endpoint del flusso di dati di destinazione.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Per specificare lo schema, è possibile creare una risorsa personalizzata schema con la definizione dello schema.

Per altre informazioni sul Registro di sistema dello schema, vedere Informazioni sugli schemi dei messaggi.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

I formati di serializzazione supportati sono JSON, Parquet e Delta.

Configurare la destinazione con un endpoint del flusso di dati per l'invio di dati

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, che sono argomenti MQTT o Kafka.

Portale

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

   

2. Selezionare Continua per configurare la destinazione.

3. Aggiungere i dettagli del mapping in base al tipo di destinazione.

Kubernetes

Ad esempio, per configurare una destinazione usando l'endpoint MQTT creato in precedenza e un argomento MQTT statico, usare la configurazione seguente:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Se sono stati creati endpoint di archiviazione come Microsoft Fabric, usare il campo destinazione dati per specificare la tabella o il nome del contenitore:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

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 dagli argomenti thermostats/+/telemetry/temperature/# MQTT e humidifiers/+/telemetry/humidity/#. La trasformazione converte la temperatura in Fahrenheit e filtra i dati in cui la temperatura è inferiore a 100000. La destinazione invia i dati all'argomento factoryMQTT .

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

Verificare che un flusso di dati funzioni

Seguire l'esercitazione: Bridge MQTT bidirezionale per 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 il portale dell'esperienza operativa o esportando la risorsa personalizzata flusso di dati.

Portale

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

Kubernetes

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