Configurar flujos de datos en operaciones de Azure IoT

Importante

En esta página se incluyen instrucciones para administrar componentes de Operaciones de IoT de Azure mediante manifiestos de implementación de Kubernetes, que se encuentra en versión preliminar. Esta característica se proporciona con varias limitacionesy no se debe usar para cargas de trabajo de producción.

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.

Importante

Cada flujo de datos debe tener el punto de conexión predeterminado del corredor MQTT local de Operaciones de IoT de Azure como origen o destino.

Puede usar la experiencia de operaciones en Operaciones de IoT de Azure para crear un flujo de datos. La experiencia de operaciones proporciona una interfaz visual para configurar el flujo de datos. También puede usar Bicep para crear un flujo de datos mediante un archivo de plantilla de Bicep o usar Kubernetes para crear un flujo de datos mediante un archivo YAML.

Siga leyendo para aprender a configurar el origen, la transformación y el destino.

Requisitos previos

Puede implementar flujos de datos en cuanto tenga una instancia de Operaciones de IoT de Azure mediante el perfil y el punto de conexión predeterminados del flujo de datos. Sin embargo, es posible que quiera configurar perfiles y puntos de conexión de flujo de datos para personalizar el flujo de datos.

Perfil de flujo de datos

Si no necesita configuraciones de escalado diferentes para los flujos de datos, use el perfil de flujo de datos predeterminado proporcionado por Operaciones de IoT de Azure. Para obtener información sobre cómo configurar un perfil de flujo de datos, consulte Configurar perfiles de flujo de datos.

Puntos de conexión de flujo de datos

Los puntos de conexión de flujo de datos son necesarios para configurar el origen y el destino del flujo de datos. Para empezar a trabajar rápidamente, puede usar el punto de conexión de flujo de datos predeterminado para el corredor MQTT local. También puede crear otros tipos de puntos de conexión de flujo de datos, 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.

Introducción

Una vez que tenga los requisitos previos, puede empezar a crear un flujo de datos.

Portal

Para crear un flujo de datos en la experiencia de operaciones, seleccione Flujo de datos>Crear flujo de datos. A continuación, verá la página donde puede configurar el origen, la transformación y el destino del flujo de datos.

Bicep

Cree un archivo de Bicep .bicep para empezar a crear un flujo de datos. En este ejemplo se muestra la estructura del flujo de datos que contiene las configuraciones de origen, transformación y destino.

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

Kubernetes (versión preliminar)

Cree un archivo de manifiesto de Kubernetes .yaml para empezar a crear un flujo de datos. En este ejemplo se muestra la estructura del flujo de datos que contiene las configuraciones de origen, transformación y destino.

apiVersion: connectivity.iotoperations.azure.com/v1
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

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

Source

Para configurar un origen para el flujo de datos, especifique la referencia del punto de conexión y una lista de orígenes de datos para el punto de conexión. Elija una de las siguientes opciones como origen para el flujo de datos.

Si el punto de conexión predeterminado no se usa como origen, debe usarse como destino. Para más información al respecto, consulte Los flujos de datos deben usar el punto de conexión del corredor MQTT local.

Opción 1: Usar el punto de conexión MQTT predeterminado como origen

Portal

1. En Detalles del origen, seleccione MQTT.

   

2. Escriba la siguiente configuración para el origen MQTT:

   Configuración	Descripción
   Tema de MQTT	Filtro de tema MQTT al que suscribirse para los mensajes entrantes. Consulte Configurar temas MQTT o Kafka.
   Esquema de los mensajes	Esquema que se va a usar para deserializar los mensajes entrantes. Consulte Especificación del esquema para deserializar datos.

3. Seleccione Aplicar.

Bicep

El punto de conexión MQTT está configurado en el archivo de plantilla de Bicep. Por ejemplo, el siguiente punto de conexión es un origen para el flujo de datos.

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

Aquí, dataSources le permite especificar varios temas MQTT o Kafka sin necesidad de modificar la configuración del punto de conexión. Esta flexibilidad significa que el mismo punto de conexión se puede reutilizar en varios flujos de datos, incluso si los temas varían. Para obtener más información, consulte Configurar orígenes de datos.

Kubernetes (versión preliminar)

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: default
  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 obtener más información, consulte Configurar orígenes de datos.

Opción 2: Usar el recurso como origen

Portal

Puede usar un recurso como origen para el flujo de datos. Usar un recurso como origen solo está disponible en la 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.

Bicep

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

Kubernetes (versión preliminar)

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

Cuando se usa un recurso como origen, la definición del recurso se usa para inferir el esquema del flujo de datos. La definición del recurso incluye el esquema de los puntos de datos del recurso. Para más información, consulte Administrar configuraciones de recursos de forma remota.

Una vez configurados, los datos del recurso alcanzaron el flujo de datos a través del corredor MQTT local. Por lo tanto, cuando se usa un recurso como origen, el flujo de datos usa el punto de conexión predeterminado del corredor MQTT local como origen en realidad.

Opción 3: Usar el punto de conexión personalizado de flujo de datos MQTT o Kafka como origen

Si creó un punto de conexión de flujo de datos MQTT o Kafka personalizado (por ejemplo, para usarlo con Event Grid o Event Hubs), puede usarlo como origen para el flujo de datos. Recuerde que los puntos de conexión de tipo de almacenamiento, como Data Lake o Fabric OneLake, no se pueden usar como origen.

Para configurar, use YAML o Bicep de Kubernetes. Reemplace los valores del marcador de posición por el nombre y los temas del punto de conexión personalizados.

Portal

Actualmente no se admite el uso de un punto de conexión MQTT o Kafka personalizado como origen en la experiencia de operaciones.

Bicep

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

Kubernetes (versión preliminar)

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

Configurar orígenes de datos (temas MQTT o Kafka)

Puede especificar varios temas de MQTT o Kafka en un origen sin necesidad de modificar la configuración del punto de conexión del flujo de datos. Esta flexibilidad significa que el mismo punto de conexión se puede reutilizar en varios flujos de datos, incluso si los temas varían. Para obtener más información, consulte Reutilizar puntos de conexión de flujo de datos.

Temas de MQTT

Cuando el origen es un punto de conexión MQTT (Event Grid incluido), puede usar el filtro de tema MQTT para suscribirse a los mensajes entrantes. El filtro de tema puede incluir caracteres comodín para suscribirse a varios temas. Por ejemplo, thermostats/+/telemetry/temperature/# se suscribe a todos los mensajes de telemetría de temperatura de termostatos. Para configurar los filtros del tema MQTT:

Portal

En los Detalles del origen del flujo de datos de experiencia de operaciones, seleccione MQTT y, a continuación, use el campo Tema MQTT para especificar el filtro de tema MQTT al que suscribirse para los mensajes entrantes.

Nota:

Solo se puede especificar un filtro de tema MQTT en la experiencia de operaciones. Para usar varios filtros de temas MQTT, use Bicep o Kubernetes.

Bicep

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

Ejemplo con varios filtros de tema MQTT con caracteres comodín:

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

Aquí se usa el carácter comodín + para seleccionar todos los dispositivos de los temas thermostats y humidifiers. El carácter comodín # se usa para seleccionar todos los mensajes de telemetría en todos los subtemas de los temas temperature y humidity.

Kubernetes (versión preliminar)

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

Ejemplo con varios filtros de tema MQTT con caracteres comodín:

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

Aquí se usa el carácter comodín + para seleccionar todos los dispositivos de los temas thermostats y humidifiers. El carácter comodín # se usa para seleccionar todos los mensajes de telemetría en todos los subtemas de los temas temperature y humidity.

Suscripciones compartidas

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

Portal

En los Detalles del origen del flujo de datos de la experiencia de operaciones, seleccione MQTT y use el campo Tema MQTT para especificar el tema y el grupo de suscripciones compartidos.

Bicep

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

Kubernetes (versión preliminar)

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

Si el recuento de instancias del perfil de flujo de datos es mayor que uno, la suscripción compartida se habilita automáticamente para todos los flujos de datos que usan el origen MQTT. En este caso, se agrega el prefijo $shared y el nombre del grupo de suscripciones compartido se genera automáticamente. Por ejemplo, si tiene un perfil de flujo de datos con un recuento de instancias de 3 y el flujo de datos usa un punto de conexión MQTT como origen configurado con los temas topic1 y topic2, se convierten automáticamente en suscripciones compartidas como $shared/<GENERATED_GROUP_NAME>/topic1 y $shared/<GENERATED_GROUP_NAME>/topic2.

Puede crear explícitamente un tema denominado $shared/mygroup/topic en la configuración. Sin embargo, no se recomienda agregar explícitamente el tema de $shared, ya que el prefijo $shared se agrega automáticamente cuando es necesario. Los flujos de datos pueden realizar optimizaciones con el nombre del grupo si no se establece. Por ejemplo, no se establece $share y los flujos de datos solo tienen que funcionar sobre el nombre del tema.

Importante

Los flujos de datos que requieren una suscripción compartida cuando el recuento de instancias es mayor que uno son importantes cuando se usa el corredor MQTT de Event Grid como origen, ya que no admite suscripciones compartidas. Para evitar que falten mensajes, establezca el recuento de instancias del perfil de flujo de datos en uno al usar el corredor MQTT de Event Grid como origen. Es cuando el flujo de datos es el suscriptor y recibe mensajes desde la nube.

Temas de Kafka

Cuando el origen es un punto de conexión Kafka (Event Hubs incluido), especifique los temas individuales de Kafka a los que suscribirse para los mensajes entrantes. No se admiten caracteres comodín, por lo que debe especificar cada tema estáticamente.

Nota:

Cuando se usa Event Hubs a través del punto de conexión Kafka, cada centro de eventos individual dentro del espacio de nombres es el tema de Kafka. Por ejemplo, si tiene un espacio de nombres de Event Hubs con dos centros de eventos, thermostats y humidifiers, puede especificar cada centro de eventos como tema de Kafka.

Para configurar los temas de Kafka:

Portal

Actualmente, no se admite el uso de un punto de conexión de Kafka como origen en la experiencia de operaciones.

Bicep

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

Kubernetes (versión preliminar)

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

Especificación del esquema de origen

Al usar MQTT o Kafka como origen, puede especificar un esquema para mostrar la lista de puntos de datos en el portal de experiencia de operaciones. Tenga en cuenta que el uso de un esquema para deserializar y validar los mensajes entrantes no se admite actualmente.

Si el origen es un recurso, el esquema se deduce automáticamente de la definición del recurso.

Sugerencia

Para generar el esquema a partir de un archivo de datos de ejemplo, use el Asistente de generación de esquemas.

Para configurar el esquema usado para deserializar los mensajes entrantes de un origen:

Portal

En los Detalles del origen del flujo de datos de la experiencia de operaciones, seleccione MQTT y use el campo Esquema de mensaje para especificar el esquema. Puede usar el botón Cargar para cargar primero un archivo de esquema. Para más información, consulte Comprender los esquemas del mensaje.

Bicep

Una vez que haya usado el registro de esquema para almacenar el esquema, puede hacer referencia a él en la configuración del flujo de datos.

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

Kubernetes (versión preliminar)

Una vez que haya usado el registro de esquema para almacenar el esquema, puede hacer referencia a él en la configuración del flujo de datos.

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

Para más información, consulte Comprender los esquemas del mensaje.

Transformación

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, Proceso, Cambiar nombreo agregar una Nueva propiedad: Mover datos de un campo a otro con una conversión opcional.

Esta sección es una introducción a las transformaciones de flujo de datos. Para obtener información más detallada, consulte Asignación de datos mediante flujos de datos, Convertir datos mediante conversiones de flujo de datos y Enriquecer datos mediante flujos de datos.

Portal

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

Bicep

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

Kubernetes (versión preliminar)

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

Enriquecimiento: adición de datos de referencia

Para enriquecer los datos, agregue primero el conjunto de datos de referencia en el almacén de estado 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.

Puede cargar datos de ejemplo en el almacén de estado mediante la CLI del almacén de estado. Los nombres de clave del almacén de estado corresponden a un conjunto de datos en la configuración del flujo de datos.

Portal

Actualmente, la fase de Enriquecer no se admite en la experiencia de operaciones.

Bicep

En este ejemplo se muestra cómo puede usar el campo deviceId en los datos de origen para que coincidan con el campo asset del conjunto de datos:

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

Kubernetes (versión preliminar)

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 de filtro y mapa.

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. Escriba la configuración necesaria.

   Configuración	Descripción
   Condición de filtro	Condición para filtrar los datos en función de un campo de los datos de origen.
   Descripción	Proporcione una descripción para la condición de filtro.

   En el campo condición de filtro, escriba @ o seleccione Ctrl + Espacio para elegir puntos de datos en una lista desplegable.

   Puede escribir las propiedades de metadatos de MQTT mediante el formato @$metadata.user_properties.<property> o @$metadata.topic. También puede escribir encabezados de $metadata mediante el formato @$metadata.<header>. La sintaxis $metadata solo es necesaria para las propiedades MQTT que forman parte del encabezado del mensaje. Para más información, consulte las referencias de campo.

   La condición puede usar los campos de los datos de origen. Por ejemplo, podría usar una condición de filtro como @temperature > 20 para filtrar datos menores o iguales a 20 en función del campo de temperatura.

3. Seleccione Aplicar.

Bicep

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

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      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.

Kubernetes (versión preliminar)

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 la experiencia de operaciones, la asignación se admite actualmente mediante transformaciones de Proceso, Cambiar nombre y Nueva propiedad.

Proceso

Puede usar la transformación Proceso para aplicar una fórmula a los datos de origen. Esta operación se usa para aplicar una fórmula a los datos de origen y almacenar el campo de resultado.

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

   

2. Escriba la configuración necesaria.

   Configuración	Descripción
   Seleccionar fórmula	Elija una fórmula existente en la lista desplegable o seleccione Personalizar para escribir una fórmula manualmente.
   Output	Especifique el nombre para mostrar de la salida para el resultado.
   Fórmula	Escriba la fórmula que se va a aplicar a los datos de origen.
   Descripción	Proporcione una descripción para la transformación.
   Último valor conocido	Opcionalmente, use el último valor conocido si el valor actual no está disponible.

   Puede escribir o editar una fórmula en el campo Fórmula. La fórmula puede usar los campos de los datos de origen. Escriba @ o seleccione Ctrl + Espacio para elegir puntos de datos en una lista desplegable.

   Puede escribir las propiedades de metadatos de MQTT mediante el formato @$metadata.user_properties.<property> o @$metadata.topic. También puede escribir encabezados de $metadata mediante el formato @$metadata.<header>. La sintaxis $metadata solo es necesaria para las propiedades MQTT que forman parte del encabezado del mensaje. Para más información, consulte las referencias de campo.

   La fórmula puede usar los campos de los datos de origen. Por ejemplo, podría usar el campo temperature en los datos de origen para convertir la temperatura a Celsius y almacenarla en el campo de salida temperatureCelsius.

3. Seleccione Aplicar.

Cambiar nombre

Puede cambiar el nombre de un punto de datos mediante la transformación Cambiar nombre. Esta operación se usa para cambiar el nombre de un punto de datos de los datos de origen a un nuevo nombre. El nuevo nombre se puede usar en las fases posteriores del flujo de datos.

1. En Transformar (opcional), seleccione Cambiar nombre>Agregar.

   

2. Escriba la configuración necesaria.

   Configuración	Descripción
   Punto de datos	Seleccione un punto de datos en la lista desplegable o escriba un encabezado $metadata.
   Nuevo nombre de punto de datos	Escriba el nuevo nombre para el punto de datos.
   Descripción	Proporcione una descripción para la transformación.

   Escriba @ o seleccione Ctrl + Espacio para elegir puntos de datos en una lista desplegable.

   Puede escribir las propiedades de metadatos de MQTT mediante el formato @$metadata.user_properties.<property> o @$metadata.topic. También puede escribir encabezados de $metadata mediante el formato @$metadata.<header>. La sintaxis $metadata solo es necesaria para las propiedades MQTT que forman parte del encabezado del mensaje. Para más información, consulte las referencias de campo.

3. Seleccione Aplicar.

Nueva propiedad

Puede agregar una nueva propiedad a los datos de origen mediante la transformación Nueva propiedad. Esta operación se usa para agregar una nueva propiedad a los datos de origen. La nueva propiedad se puede usar en las fases posteriores del flujo de datos.

1. En Transformar (opcional), seleccione Nueva propiedad>Agregar.

   

2. Escriba la configuración necesaria.

   Configuración	Descripción
   Clave de propiedad	Escriba la clave de la nueva propiedad.
   Valor de propiedad	Escriba el valor de la nueva propiedad.
   Descripción	Proporcione una descripción para la nueva propiedad.

3. Seleccione Aplicar.

Bicep

Puede acceder a las propiedades de metadatos de MQTT mediante el formato $metadata.user_properties.<property> o $metadata.topic. También puede escribir encabezados de $metadata mediante el formato $metadata.<header>. Para más información, consulte las referencias de campo.

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

Kubernetes (versión preliminar)

Puede acceder a las propiedades de metadatos de MQTT mediante el formato $metadata.user_properties.<property> o $metadata.topic. También puede escribir encabezados de $metadata mediante el formato $metadata.<header>. Para más información, consulte las referencias de campo.

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
      expression: "($1 - 32) * 5/9"
      output: temperatureCelsius
    - 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. Los puntos de conexión de almacenamiento como Microsoft Fabric o Azure Data Lake requieren un esquema para garantizar la coherencia de los datos. Los formatos de serialización admitidos son Parquet y Delta.

Sugerencia

Para generar el esquema a partir de un archivo de datos de ejemplo, use el Asistente de generación de esquemas.

Portal

Para la experiencia de operaciones, especifique el formato de esquema y serialización en los detalles del punto de conexión del flujo de datos. Los puntos de conexión que admiten formatos de serialización son Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2 y Azure Data Explorer. Por ejemplo, para serializar los datos en formato Delta, debe cargar un esquema en el registro de esquemas y hacer referencia a ellos en la configuración del punto de conexión de destino del flujo de datos.

Bicep

Una vez cargado un esquema en el registro de esquemas, puede hacer referencia a él en la configuración del flujo de datos.

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

Kubernetes (versión preliminar)

Una vez cargado un esquema en el registro de esquemas, puede hacer referencia a él en la configuración del flujo de datos.

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

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

Destino

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.

Para enviar datos a un destino distinto del corredor MQTT local, cree un punto de conexión de flujo de datos. Para obtener información sobre cómo hacerlo, consulte Configurar puntos de conexión de flujo de datos. Si el destino no es el corredor MQTT local, debe usarse como origen. Para más información al respecto, consulte Los flujos de datos deben usar el punto de conexión del corredor MQTT local.

Importante

Los puntos de conexión de almacenamiento requieren un esquema para la serialización. Para usar el flujo de datos con Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer o Local Storage, debe especificar una referencia de esquema.

Portal

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

   

   Los puntos de conexión de almacenamiento requieren un esquema para la serialización. Si elige un punto de conexión de destino de Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer o de almacenamiento local, debe especificar una referencia de esquema. Por ejemplo, para serializar los datos en un punto de conexión de Microsoft Fabric en formato Delta, debe cargar un esquema en el registro de esquemas y hacer referencia a este en la configuración del punto de conexión de destino del flujo de datos.

   

2. Seleccione Continuar para configurar el destino.

3. Escriba la configuración necesaria para el destino, incluido el tema o la tabla a los que enviar los datos. Consulte Configurar destino de datos (tema, contenedor o tabla) para obtener más información.

Bicep

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

Kubernetes (versión preliminar)

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

Configurar el destino de datos (tema, contenedor o tabla)

De forma similar a los orígenes de datos, el destino de datos es un concepto que se usa para mantener los puntos de conexión de flujo de datos reutilizables en varios flujos de datos. Básicamente, representa el subdirectorio en la configuración del punto de conexión del flujo de datos. Por ejemplo, si el punto de conexión de flujo de datos es un punto de conexión de almacenamiento, el destino de datos es la tabla de la cuenta de almacenamiento. Si el punto de conexión de flujo de datos es un punto de conexión de Kafka, el destino de datos es el tema de Kafka.

Tipo de punto de conexión	Significado del destino de datos	Descripción
MQTT (o Event Grid)	Tema	El tema MQTT donde se envían los datos. Solo se admiten temas estáticos, sin caracteres comodín.
Kafka (o Event Hubs)	Tema	Tema de Kafka donde se envían los datos. Solo se admiten temas estáticos, sin caracteres comodín. Si el punto de conexión es un espacio de nombres de Event Hubs, el destino de datos es el centro de eventos individual dentro del espacio de nombres.
Almacén de Azure Data Lake	Contenedor	Contenedor de la cuenta de almacenamiento. No la tabla.
Microsoft Fabric OneLake	Tabla o carpeta	Corresponde al tipo de ruta de acceso para el punto de conexión configurado.
Explorador de datos de Azure	Tabla	Tabla de la base de datos de Azure Data Explorer.
Almacenamiento local	Carpeta	Nombre de carpeta o directorio en el montaje del volumen persistente de almacenamiento local. Al usar el Almacenamiento de contenedores de Azure habilitado por volúmenes perimetrales de ingesta en la nube de Azure Arc, esto debe coincidir con el parámetro spec.path para el subvolumen que creó.

Para configurar el destino de datos:

Portal

Al usar la experiencia de operaciones, el campo de destino de datos se interpreta automáticamente en función del tipo de punto de conexión. Por ejemplo, si el punto de conexión de flujo de datos es un punto de conexión de almacenamiento, la página de detalles de destino le pedirá que escriba el nombre del contenedor. Si el punto de conexión de flujo de datos es un punto de conexión MQTT, la página de detalles de destino le pedirá que escriba el tema, etc.

Bicep

La sintaxis es la misma para todos los puntos de conexión de flujo de datos:

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

Por ejemplo, para enviar datos de vuelta al corredor MQTT local utilizando un tema MQTT estático, use la siguiente configuración:

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

O bien, si tiene un punto de conexión del centro de eventos personalizado, la configuración tendría el siguiente aspecto:

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

Otro ejemplo que usa un punto de conexión de almacenamiento como destino:

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

Kubernetes (versión preliminar)

La sintaxis es la misma para todos los puntos de conexión de flujo de datos:

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

Por ejemplo, para enviar datos de vuelta al corredor MQTT local utilizando un tema MQTT estático, use la siguiente configuración:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

O bien, si tiene un punto de conexión del centro de eventos personalizado, la configuración tendría el siguiente aspecto:

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

Otro ejemplo que usa un punto de conexión de almacenamiento como destino:

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

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 del tema azure-iot-operations/data/thermostat de MQTT. La transformación convierte la temperatura a Fahrenheit y filtra los datos donde la temperatura multiplicada por la humedad sea inferior a 100000. El destino envía los datos al tema MQTT factory.

Portal

Consulte las pestañas Bicep o Kubernetes para ver el ejemplo de configuración.

Bicep

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 dataflow endpoint
resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow 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 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'
        }
      }
    ]
  }
}

Kubernetes (versión preliminar)

apiVersion: connectivity.iotoperations.azure.com/v1
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

Para ver más ejemplos de configuraciones de flujo de datos, consulte API REST de Azure: flujo de datos y el inicio rápido de Bicep.

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 la experiencia de operaciones o exportar el recurso personalizado de flujo de datos.

Portal

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

Bicep

Bicep es la infraestructura como código y no se requiere ninguna exportación. Use el archivo de plantilla de Bicep para crear un flujo de datos para establecer y configurar rápidamente flujos de datos.

Kubernetes (versión preliminar)

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

Configuración correcta del flujo de datos

Para asegurarse de que el flujo de datos funciona según lo previsto, compruebe lo siguiente:

• El punto de conexión de flujo de datos MQTT predeterminado debe usarse como origen o destino.
• El perfil de flujo de datos existe y se le hace referencia en la configuración del flujo de datos.
• El origen es un punto de conexión MQTT, un punto de conexión de Kafka o un recurso. Los puntos de conexión de tipo de almacenamiento no se pueden usar como origen.
• Cuando se usa Event Grid como origen, el recuento de instancias del perfil de flujo de datos se establece en 1 porque el corredor MQTT de Event Grid no admite suscripciones compartidas.
• Cuando se usa Event Hubs como origen, cada centro de eventos del espacio de nombres es un tema de Kafka independiente y debe especificarse como origen de datos.
• La transformación, si se usa, se configura con la sintaxis correcta, incluida la secuencia de caracteres de escape apropiada en los caracteres especiales.
• Cuando se usan puntos de conexión de tipo de almacenamiento como destino, se especifica un esquema.

Pasos siguientes

• Asignar datos mediante flujos de datos
• Convertir datos mediante flujos de datos
• Enriquecer datos mediante flujos de datos
• Descripción de los esquemas de mensajes
• Administrar perfiles de flujo de datos