Configurar flujos de datos en operaciones de Azure IoT

Artículo
11/05/2024

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.

Tendrá que implementar una nueva instalación de Operaciones de IoT de Azure cuando esté disponible una versión disponible con carácter general. No podrá actualizar una instalación de versión preliminar.

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, consulte los Términos de uso complementarios para las versiones preliminares de Microsoft Azure.

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.

Diagrama de un flujo de datos que muestra el flujo desde el origen hasta 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 (versión preliminar) 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

El perfil de flujo de datos especifica el número de instancias de los flujos de datos que se van a usar. Si no necesita varios grupos de flujos de datos con diferentes configuraciones de escalado, puede usar el perfil de flujo de datos predeterminado. 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.

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.

Recorte de pantalla que usa la experiencia de operaciones para crear un flujo de datos.

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

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/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

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.

Usar el recurso como origen

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

En Detalles del origen, seleccione Recurso.
Seleccione el recurso que desea usar como punto de conexión de origen.
Seleccione Continuar.

Se muestra una lista de puntos de datos para el recurso seleccionado.
Seleccione Aplicar para usar el recurso como punto de conexión de origen.

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.

Usar punto de conexión MQTT predeterminado como origen

En Detalles del origen, seleccione MQTT.

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.

Seleccione Aplicar.

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.

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.

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.

Usar punto de conexión de flujo de datos MQTT o Kafka personalizado 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.

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

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

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:

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.

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.

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/#

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

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.

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

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

Si el recuento de instancias del perfil de flujo de datos es mayor que 1, 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. Si desea usar un identificador diferente del grupo de suscripciones compartido, puede invalidarlo en el tema, como $shared/mygroup/topic1.

Importante

Los flujos de datos que requieren una suscripción compartida cuando el recuento de instancias es mayor que 1 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 1 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:

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

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

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.

Especificar el esquema solo es relevante cuando se usa el origen MQTT o Kafka. Si el origen es un recurso, el esquema se deduce automáticamente de la definición del recurso.

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

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.

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>'
}

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>'

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:

Enriquecer, Cambiar nombre o agregar una Propiedad nueva: agregue datos adicionales a los datos de origen dados un conjunto de datos y una condición para que coincidan.
Filtrar: filtre los datos en función de una condición.
Asignar o Procesar: mover datos de un campo a otro con una conversión opcional.

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

Recorte de pantalla que usa la experiencia de operaciones para agregar una transformación a un flujo de datos.

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

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.

Puede cargar datos de muestra en el DSS usando la herramienta de configuración de DSS de muestra. Los nombres de clave del almacén de estado distribuido corresponden a un conjunto de datos en la configuración del flujo de datos.

En la experiencia de operaciones, actualmente se admite la fase de Enriquecer mediante las transformaciones Cambiar nombre y Nueva propiedad.

En la experiencia de operaciones, seleccione un flujo de datos y luego Agregar transformación (opcional).
Elija las transformaciones Cambiar nombre o Nueva propiedad y, a continuación, seleccione Agregar.

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

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.

En Transformar (opcional), seleccione Filtrar>Agregar.
Elija los puntos de datos que se van a incluir en el conjunto de datos.
Agregue una condición de filtro y una descripción.
Seleccione Aplicar.

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.

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.

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.

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

En Transformar (opcional), seleccione Compute>Agregar.
Escriba los campos y expresiones necesarios.
Seleccione Aplicar.

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

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.

Actualmente, no se admite la especificación del esquema de salida y la serialización en la experiencia de operaciones.

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>'
}

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 una referencia de esquema. Si ha creado puntos de conexión de destino de almacenamiento para Microsoft Fabric OneLake, ADLS Gen 2, Azure Data Explorer y Almacenamiento local, debe especificar la referencia del esquema.

Seleccione el punto de conexión de flujo de datos que se va a usar como destino.
Seleccione Continuar para configurar el destino.
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.

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

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:

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.

Recorte de pantalla que muestra la experiencia de operaciones que pide al usuario que escriba un tema MQTT según el tipo de punto de conexión.

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'
}

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 en Fahrenheit y filtra los datos donde la temperatura multiplicada por la humedad es inferior a 100 000. El destino envía los datos al tema MQTT factory.

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

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

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.

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

Recorte de pantalla que usa la experiencia de operaciones para exportar un flujo de datos.

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.

Compartir a través de

Configurar flujos de datos en operaciones de Azure IoT

Requisitos previos

Perfil de flujo de datos

Puntos de conexión de flujo de datos

Introducción

Source

Usar el recurso como origen

Usar punto de conexión MQTT predeterminado como origen

Usar punto de conexión de flujo de datos MQTT o Kafka personalizado como origen

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

Temas de MQTT

Suscripciones compartidas

Temas de Kafka

Especificación del esquema para deserializar datos

Transformación

Enriquecimiento: adición de datos de referencia

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

Asignación: mover datos de un campo a otro

Serializar datos según un esquema

Destino

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

Ejemplo

Comprobación de que un flujo de datos funciona

Exportar la configuración del flujo de datos

Configuración correcta del flujo de datos

Pasos siguientes

Comentarios

Recursos adicionales