Configurar flujos de datos en operaciones de Azure IoT

Importante

Versión preliminar de operaciones de Azure IoT: habilitada por Azure Arc está actualmente en versión preliminar. No se debería usar este software en versión preliminar en entornos de producción.

Deberá implementar una nueva instalación de Azure IoT Operations cuando esté disponible una versión general. No podrá actualizar una instalación de versión preliminar.

Consulte Términos de uso complementarios para las versiones preliminares de Microsoft Azure para conocer los términos legales que se aplican a las características de Azure que se encuentran en la versión beta, en versión preliminar o que todavía no se han publicado para que estén disponibles con carácter general.

Un flujo de datos es la ruta de acceso que los datos toman del origen al destino con transformaciones opcionales. Puede configurar el flujo de datos creando un recurso personalizado del flujo de datos o utilizando el portal de Azure IoT Operations Studio. Un flujo de datos se compone de tres partes: el origen, la transformación y el destino.

Para definir el origen y el destino, debe configurar los puntos de conexión de flujo de datos. La transformación es opcional y puede incluir operaciones como enriquecer los datos, filtrar los datos y asignar los datos a otro campo.

En este artículo se muestra cómo crear un flujo de datos con un ejemplo, incluido el origen, la transformación y el destino.

Requisitos previos

• Una instancia de la versión preliminar de Operaciones de IoT de Azure
• Un perfil de flujo de datos configurado
• Puntos de conexión de flujo de datos. Por ejemplo, cree un punto de conexión de flujo de datos para el broker MQTT local. Puede usar este punto de conexión para el origen y el destino. O bien, puede probar otros puntos de conexión como Kafka, Event Hubs o Azure Data Lake Storage. Para obtener información sobre cómo configurar cada tipo de punto de conexión de flujo de datos, consulte Configuración de puntos de conexión de flujo de datos.

Crear flujo de datos

Una vez que tenga puntos de conexión de flujo de datos, puede usarlos para crear un flujo de datos. Recuerde que un flujo de datos se compone de tres partes: el origen, la transformación y el destino.

Portal

Para crear un flujo de datos en el portal de experiencia de operaciones, seleccione Flujo de datos>Crear flujo de datos.

Kubernetes

La estructura general de una configuración de flujo de datos es la siguiente:

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

Revise las secciones siguientes para obtener información sobre cómo configurar los tipos de operación del flujo de datos.

Configurar un origen con un punto de conexión de flujo de datos para obtener datos

Para configurar un origen para el flujo de datos, especifique la referencia del punto de conexión y el origen de datos. Puede especificar una lista de orígenes de datos para el punto de conexión.

Portal

Usar activo como origen

Puede usar un recurso como origen para el flujo de datos. Esto solo está disponible en el portal de experiencia de operaciones.

1. En Detalles del origen, seleccione Recurso.

2. Seleccione el recurso que desea usar como punto de conexión de origen.

3. Seleccione Continuar.

   Se muestra una lista de puntos de datos para el recurso seleccionado.

   

4. Seleccione Aplicar para usar el recurso como punto de conexión de origen.

Kubernetes

La configuración de un recurso como origen solo está disponible en el portal de experiencia de operaciones.

Usar de MQTT como origen

Portal

1. En Detalles del origen, seleccione MQTT.

2. Escriba el tema de MQTT al que desea escuchar los mensajes entrantes.

3. Elija un esquema de mensaje en la lista desplegable o cargue un nuevo esquema. Si los datos de origen tienen campos o campos opcionales con tipos diferentes, especifique un esquema de deserialización para garantizar la coherencia. Por ejemplo, los datos pueden tener campos que no están presentes en todos los mensajes. Sin el esquema, la transformación no puede controlar estos campos, ya que tendría valores vacíos. Con el esquema, puede especificar valores predeterminados o omitir los campos.

   

4. Seleccione Aplicar.

Kubernetes

Por ejemplo, para configurar un origen mediante un punto de conexión MQTT y dos filtros de tema MQTT, use la siguiente configuración:

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

Dado que dataSources permite especificar temas MQTT o Kafka sin modificar la configuración del punto de conexión, puede reutilizar el punto de conexión para varios flujos de datos incluso si los temas son diferentes. Para más información, consulte Reutilizar puntos de conexión de flujo de datos.

Especificación del esquema para deserializar datos

Si los datos de origen tienen campos o campos opcionales con tipos diferentes, especifique un esquema de deserialización para garantizar la coherencia. Por ejemplo, los datos pueden tener campos que no están presentes en todos los mensajes. Sin el esquema, la transformación no puede controlar estos campos, ya que tendría valores vacíos. Con el esquema, puede especificar valores predeterminados o omitir los campos.

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

Para especificar el esquema, cree el archivo y almacénelo en el registro de esquema.

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

Nota:

El único formato de serialización admitido es JSON. El esquema es opcional.

Para obtener más información sobre el registro de esquemas, consulte Comprender los esquemas de mensajes.

Suscripciones compartidas

Para usar suscripciones compartidas con orígenes MQTT, puede especificar el tema de suscripción compartida en forma de $shared/<subscription-group>/<topic>.

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

Nota:

Si el recuento de instancias del perfil de flujo de datos es mayor que 1, se debe usar el tema de la suscripción compartida.

Configurar la transformación para procesar datos

La operación de transformación es donde puede transformar los datos del origen antes de enviarlos al destino. Las transformaciones son opcionales. Si no necesita realizar cambios en los datos, no incluya la operación de transformación en la configuración del flujo de datos. Varias transformaciones se encadenan en fases independientemente del orden en que se especifiquen en la configuración. El orden de las fases siempre es

1. Enriquecimiento: agregue datos adicionales a los datos de origen dados un conjunto de datos y una condición para que coincidan.
2. Filtrar: filtre los datos en función de una condición.
3. Asignar: mover datos de un campo a otro con una conversión opcional.

Portal

En el portal de experiencia de operaciones, seleccione Flujo de datos>Agregar transformación (opcional).

Kubernetes

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

Enriquecimiento: adición de datos de referencia

Para enriquecer los datos, puede usar el conjunto de datos de referencia en el almacén de estado distribuido (DSS) de Operaciones de IoT de Azure. El conjunto de datos se usa para agregar datos adicionales a los datos de origen en función de una condición. La condición se especifica como un campo en los datos de origen que coincide con un campo del conjunto de datos.

Los nombres de clave del almacén de estado distribuido corresponden a un conjunto de datos en la configuración del flujo de datos.

Portal

Actualmente, la operación de enriquecimiento no está disponible en el portal de experiencia de operaciones.

Kubernetes

Por ejemplo, puede usar el campo deviceId de los datos de origen para que coincida con el campo asset del conjunto de datos:

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

Si el conjunto de datos tiene un registro con el campo asset, similar a:

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

Los datos del origen con el campo deviceId que coincide con thermostat1 tienen los campos location y manufacturer disponibles en las fases filter y map.

Puede cargar datos de muestra en el DSS usando la herramienta de configuración de DSS de muestra.

Para más información sobre la sintaxis de las condiciones, consulte Enriquecimiento de datos mediante flujos de datos y Conversión de datos mediante flujos de datos.

Filtro: Filtrar datos en función de una condición

Para filtrar los datos en una condición, puede usar la fase filter. La condición se especifica como un campo en los datos de origen que coincide con un valor.

Portal

1. En Transformar (opcional), seleccione Filtrar>Agregar.

2. Elija los puntos de datos que se van a incluir en el conjunto de datos.

3. Agregue una condición de filtro y una descripción.

   

4. Seleccione Aplicar.

Kubernetes

Por ejemplo, podría usar el campo temperature de los datos de origen para filtrar los datos:

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

Si el campo temperature es mayor que 20, los datos se pasan a la siguiente fase. Si el campo temperature es menor o igual que 20, se filtran los datos.

Asignación: mover datos de un campo a otro

Para asignar los datos a otro campo con conversión opcional, puede usar la operación map. La conversión se especifica como una fórmula que usa los campos de los datos de origen.

Portal

En el portal de experiencia de operaciones, la asignación se admite actualmente mediante transformaciones de Compute.

1. En Transformar (opcional), seleccione Compute>Agregar.

2. Escriba los campos y expresiones necesarios.

   

3. Seleccione Aplicar.

Kubernetes

Por ejemplo, podría usar el campo temperature en los datos de origen para convertir la temperatura a Celsius y almacenarla en el campo temperatureCelsius. También puede enriquecer los datos de origen con el campo location del conjunto de datos de contextualización:

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

Para más información, consulte Asignación de datos mediante flujos de datos y Convertir datos mediante flujos de datos.

Serializar datos según un esquema

Si desea serializar los datos antes de enviarlos al destino, debe especificar un esquema y un formato de serialización. De lo contrario, los datos se serializan en JSON con los tipos inferidos. Recuerde que los puntos de conexión de almacenamiento como Microsoft Fabric o Azure Data Lake requieren un esquema para garantizar la coherencia de los datos.

Portal

Especifique el esquema de salida cuando agregue el punto final del flujo de datos de destino.

Kubernetes

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

Para especificar el esquema, puede crear un recurso personalizado de esquema con la definición de esquema.

Para obtener más información sobre el registro de esquemas, consulte Comprender los esquemas de mensajes.

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

Los formatos de serialización admitidos son JSON, Parquet y Delta.

Configurar el destino con un punto de conexión de flujo de datos para enviar datos

Para configurar un destino para el flujo de datos, especifique la referencia del punto de conexión y el destino de datos. Puede especificar una lista de destinos de datos para el punto de conexión que son temas MQTT o Kafka.

Portal

1. Seleccione el punto de conexión de flujo de datos que se va a usar como destino.

   

2. Seleccione Continuar para configurar el destino.

3. Agregue los detalles de asignación en función del tipo de destino.

Kubernetes

Por ejemplo, para configurar un destino mediante el punto de conexión MQTT creado anteriormente y un tema de MQTT estático, use la siguiente configuración:

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Si ha creado puntos de conexión de almacenamiento como Microsoft Fabric, use el campo de destino de datos para especificar la tabla o el nombre del contenedor:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

Ejemplo

El ejemplo siguiente es una configuración de flujo de datos que usa el punto de conexión MQTT para el origen y el destino. El origen filtra los datos de los temas thermostats/+/telemetry/temperature/# de MQTT y humidifiers/+/telemetry/humidity/#. La transformación convierte la temperatura en Fahrenheit y filtra los datos donde la temperatura es inferior a 100000. El destino envía los datos al tema MQTT factory.

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

Comprobación de que un flujo de datos funciona

Siga el tutorial: Puente MQTT bidireccional a Azure Event Grid para comprobar que el flujo de datos funciona.

Exportar la configuración del flujo de datos

Para exportar la configuración del flujo de datos, puede usar el portal de experiencia de operaciones o exportando el recurso personalizado de flujo de datos.

Portal

Seleccione el flujo de datos que desea exportar y seleccione Exportar en la barra de herramientas.

Kubernetes

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