Creación de flujos de datos en Azure IoT Operations

1. Para crear un data flow en operations experience, seleccione Data flow>Crear data flow.

2. Seleccione el marcador de posición new-data-flow para establecer las propiedades del flujo de datos. Escriba el nombre del data flow y elija el perfil de data flow que se va a usar. El perfil de flujo de datos predeterminado está seleccionado de forma predeterminada. Para obtener más información sobre los perfiles de data flow, consulte Configure data flow profile.

   

   Importante

   Solo puedes elegir el perfil del flujo de datos al crear un flujo de datos. No se puede cambiar el perfil de flujo de datos después de que el flujo de datos ha sido creado. Si desea cambiar el perfil de data flow de un data flow existente, elimine el data flow original y cree uno nuevo con el nuevo perfil de data flow.

3. Configure el punto de conexión de origen, transformación y destino para el flujo de datos seleccionando los elementos del diagrama de flujo de datos.

   

Use el comando az iot ops dataflow apply para crear o cambiar un data flow.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

El --config-file parámetro es la ruta de acceso y el nombre de archivo de un archivo de configuración JSON que contiene las propiedades del recurso.

En este ejemplo, supongamos que se almacena un archivo de configuración denominado data-flow.json con el siguiente contenido en el directorio principal del usuario:

{
    "mode": "Enabled",
    "operations": [
        {
            "operationType": "Source",
            "sourceSettings": {
                // See source configuration section
            }
        },
        {
            "operationType": "BuiltInTransformation",
            "builtInTransformationSettings": {
                // See transformation configuration section
            }
        },
        {
            "operationType": "Destination",
            "destinationSettings": {
                // See destination configuration section
            }
        }
    ]
}

Este es un comando de ejemplo para crear o actualizar un data flow mediante el perfil de flujo de datos predeterminado:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

Cree un archivo Bicep .bicep para empezar a crear un flujo de datos. En este ejemplo se muestra la estructura del data flow 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 data flow 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 data flow 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 comenzar a crear un flujo de datos. En este ejemplo se muestra la estructura del data flow 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 data flow 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

1. En Detalles del origen, seleccione Agente de mensajes.

   

2. Escriba la siguiente configuración para el origen del agente de mensajes:

   Configuración	Descripción
   punto de conexión de flujo de datos	Seleccione Predeterminado para usar el punto de conexión predeterminado del agente de mensajes de MQTT.
   Tema	Filtro de tema al que suscribirse para los mensajes entrantes. Use Temas>Agregar una fila para agregar varios temas. Para obtener más información sobre los temas, consulte Configuración de MQTT o Temas de 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.

Esta es una configuración de ejemplo del punto de conexión de origen para el corredor MQTT predeterminado.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "thermostats/+/sensor/temperature/#",
      "humidifiers/+/sensor/humidity/#"
    ],
    "endpointRef": "default"
  }
}

Configura el punto de conexión del broker de mensajes en el archivo de Bicep. Por ejemplo, el siguiente punto de conexión es una fuente para el flujo de datos.

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

Para configurar un origen que use un punto de conexión del agente de mensajes y dos filtros de tema, use la siguiente configuración:

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

Puede usar un asset como origen del data flow. Puede usar un recurso como fuente solo 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.

Puede configurar un recurso como origen solo dentro de la interfaz de operaciones.

Puede configurar un recurso como origen solo dentro de la interfaz de operaciones.

Puede configurar un recurso como origen solo dentro de la interfaz de operaciones.

1. En Detalles del origen, seleccione Agente de mensajes.

   

2. Escriba la siguiente configuración para el origen del agente de mensajes:

   Configuración	Descripción
   punto de conexión de flujo de datos	Use el botón Reselect para seleccionar un punto de conexión personalizado de MQTT o Kafka data flow. Para obtener más información, consulte Configurar los puntos de conexión de flujo de datos MQTT o Configurar Azure Event Hubs y los puntos de conexión de flujo de datos Kafka.
   Tema	Filtro de tema al que suscribirse para los mensajes entrantes. Use Temas>Agregar una fila para agregar varios temas. Para obtener más información sobre los temas, consulte Configuración de MQTT o Temas de 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.

Reemplace los valores del marcador de posición por el nombre y los temas del punto de conexión personalizados.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<CUSTOM_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_1>",
            "<TOPIC_2>"
        ]
    }
}

Reemplace los valores del marcador de posición por el nombre y los temas del punto de conexión personalizados.

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

Reemplace los valores del marcador de posición por el nombre y los temas del punto de conexión personalizados.

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

En el flujo de datos de la experiencia de operaciones Source details, seleccione Message broker y use el campo Topic(s) para especificar los filtros de tópicos MQTT a los que suscribirse de los mensajes entrantes. Para agregar varios temas mqTT, seleccione Agregar fila y escriba un nuevo tema.

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "<MESSAGE_BROKER_ENDPOINT_NAME>",
        "dataSources": [
            "<TOPIC_FILTER_1>",
            "<TOPIC_FILTER_2>"
            // Add more topic filters as needed
        ]
    }
}

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

{
    "operationType": "Source",
    "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
        ]
    }
}

Aquí, el carácter comodín + selecciona todos los dispositivos bajo los temas thermostats y humidifiers. El carácter comodín # selecciona todos los mensajes del sensor en todos los subtemas de los temas temperature y humidity.

sourceSettings: {
  endpointRef: '<MESSAGE_BROKER_ENDPOINT_NAME>'
  dataSources: [
    '<TOPIC_FILTER_1>'
    '<TOPIC_FILTER_2>'
    // Add more topic filters as needed
  ]
}

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

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

Aquí, el carácter comodín + selecciona todos los dispositivos bajo los temas thermostats y humidifiers. El carácter comodín # selecciona todos los mensajes del sensor en todos los subtemas de los temas temperature y humidity.

sourceSettings:
  endpointRef: <ENDPOINT_NAME>
  dataSources:
    - <TOPIC_FILTER_1>
    - <TOPIC_FILTER_2>
    # Add more topic filters as needed

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

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

Aquí, el carácter comodín + selecciona todos los dispositivos bajo los temas thermostats y humidifiers. El carácter comodín # selecciona todos los mensajes en todos los subtemas de los temas temperature y humidity.

En el flujo de datos de experiencia en operaciones Source details, seleccione Message broker y use el campo Topic para especificar el grupo de suscripciones compartido y el tópico.

{
  "operationType": "Source",
  "sourceSettings": {
    "dataSources": [
      "$shared/<GROUP_NAME>/<TOPIC_FILTER>"
    ]
  }
}

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

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

En el flujo de datos de la experiencia de operaciones Source details, seleccione Message broker, y luego use el campo Topic para especificar el filtro de tema de Kafka para suscribirse a los mensajes entrantes.

{
  "operationType": "Source",
  "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
  ]
}

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

En el flujo de datos de la experiencia de operaciones Source details, seleccione Message broker y use el campo Message schema para especificar el esquema. Seleccione Cargar para cargar un archivo de esquema. Para obtener más información, consulte Descripción de los esquemas de mensajes.

{
  "operationType": "Source",
  "sourceSettings": {
    "endpointRef": "<ENDPOINT_NAME>",
    "serializationFormat": "Json",
    "schemaRef": "aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>"
  }
}

Después de usar el registro schema para almacenar el esquema, haga referencia a él en la configuración de data flow.

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

Después de usar el registro schema para almacenar el esquema, haga referencia a él en la configuración de data flow.

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

Al crear o editar un data flow, seleccione Edit y, a continuación, seleccione Yes junto a Request data persistence.

Agregue la propiedad requestDiskPersistence al archivo de configuración de data flow:

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

Agregue la propiedad requestDiskPersistence al recurso de flujo de datos. La versión de la API es 2025-10-01 o posterior:

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

Agregue la propiedad requestDiskPersistence a la especificación de data flow. Use la versión de API connectivity.iotoperations.azure.com/v1beta1 o posterior:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

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

{
  "operationType": "BuiltInTransformation",
  "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
  ]
}

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

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

Para enriquecer los datos, use la propiedad builtInTransformationSettings en la configuración de data flow. Use la datasets propiedad para especificar los conjuntos de datos para el enriquecimiento.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "datasets": [
      {
        "key": "<DATASET_KEY>",
        "inputs": [
          "$source.<SOURCE_FIELD>" // ---------------- $1
          "$context(<DATASET_KEY>).<DATASET_FIELD>" // - $2
        ],
        "expression": "$1 == $2"
      }
    ]
  }
}

En este ejemplo se muestra cómo puede usar el deviceId campo de los datos de origen para que coincidan con el asset campo 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

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.

   Escriba las propiedades de metadatos de MQTT mediante el formato @$metadata.user_properties.<property> o @$metadata.topic. Ingrese los encabezados de $metadata utilizando 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, use una condición de filtro como @temperature > 20 para filtrar datos menores o iguales que 20 en función del campo de temperatura.

3. Seleccione Aplicar.

Por ejemplo, use el temperature campo de los datos de origen para filtrar los datos:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "filter": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "expression": "$1 > 20"
      }
    ]
  }
}

Por ejemplo, use el temperature campo 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, use el temperature campo 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.

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.

Por ejemplo, puede usar el temperature campo de los datos de origen para convertir la temperatura a Celsius y almacenarlo en el temperatureCelsius campo. Enriquecer los datos de origen con el location campo del conjunto de datos de contextualización:

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "$source.temperature ? $last" // ---------------- $1
        ],
        "output": "temperatureCelsius",
        "expression": "($1 - 32) * 5/9"
      },
      {
        "inputs": [
          "$context(assetDataset).location" // - $2
        ],
        "output": "location"
      }
    ]
  }
}

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

Por ejemplo, puede usar el temperature campo de los datos de origen para convertir la temperatura a Celsius y almacenarlo en el temperatureCelsius campo. Enriquecer los datos de origen con el location campo del conjunto de datos de contextualización:

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

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

Por ejemplo, puede usar el temperature campo de los datos de origen para convertir la temperatura a Celsius y almacenarlo en el temperatureCelsius campo. Enriquecer los datos de origen con el location campo 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

1. En Transformar (opcional), seleccione Quitar.

2. Seleccione el punto de datos que se va a quitar del esquema de salida.

   

3. Seleccione Aplicar.

Para quitar un punto de datos del esquema de salida, use la propiedad builtInTransformationSettings en la configuración de data flow. Use la map propiedad para especificar los puntos de datos que se van a quitar.

{
  "operationType": "BuiltInTransformation",
  "builtInTransformationSettings": {
    "map": [
      {
        "inputs": [
          "*"
        ],
        "output": "*"
      },
      {
        "inputs": [
          "weight"
        ],
        "output": ""
      }
      {
          "inputs": [
          "weight.SourceTimestamp"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.Value"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Code"
          ],
          "output": ""
      },
      {
          "inputs": [
          "weight.StatusCode.Symbol"
          ],
          "output": ""
      }
    ]
  }
}

builtInTransformationSettings: {
  map: [
    {
      inputs: [
        '*'
      ]
      output: '*'
    }
    {
      inputs: [
        'weight'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.SourceTimestamp'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.Value'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Code'
      ]
      output: ''
    }
    {
      inputs: [
        'weight.StatusCode.Symbol'
      ]
      output: ''
    }
  ]
}

builtInTransformationSettings:
  map:
    - type: PassThrough
      inputs:
        - "*"
      output: "*"
    - inputs:
        - weight
      output: ""
    - inputs:
        - weight.SourceTimestamp
      output: ""
    - inputs:
        - weight.Value
      output: ""
    - inputs:
        - weight.StatusCode
      output: ""
    - inputs:
        - weight.StatusCode.Code
      output: ""
    - inputs:
        - weight.StatusCode.Symbol
      output: ""

Para la experiencia de Operaciones, especifique el esquema y el formato de serialización en los detalles del punto de conexión de flujo de datos. Los puntos de conexión que admiten formatos de serialización son Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Azure Data Explorer y storage locales. Por ejemplo, para serializar los datos en formato Delta, cargue un esquema en el registro de esquemas y haga referencia a él en la configuración del punto de conexión de destino de data flow.

Después de cargar un esquema en el Registro de esquemas, haga referencia a él en la configuración de data flow.

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

Después de cargar un esquema en el Registro de esquemas, haga referencia a él en la configuración de data flow.

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

Después de cargar un esquema en el Registro de esquemas, haga referencia a él en la configuración de data flow.

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

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

   

   los puntos de conexión de Storage requieren un schema 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 Local Storage, 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.

{
  "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
}

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

Cuando se usa la experiencia de operaciones, interpreta automáticamente el campo de destino de datos 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 del destino le pedirá que escriba el nombre del contenedor. Si el punto de conexión del flujo de datos es un punto de conexión MQTT, la página de detalles de destino le solicitará que introduzca el tema, etc.

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

Para devolver datos al agente MQTT local, use un tema de MQTT estático en la siguiente configuración:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "example-topic"
  }
}

Si tiene un punto de conexión de centro de eventos personalizado, la configuración es similar a la siguiente:

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

En el caso de los puntos de conexión MQTT, también puede usar variables de tema dinámicos. Para enrutar mensajes de factory/1/data a processed/factory/1, use el ejemplo siguiente:

{
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "processed/factory/${inputTopic.2}"
  }
}

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

Para devolver datos al agente MQTT local, use un tema de MQTT estático en la siguiente configuración:

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

Si tiene un punto de conexión de centro de eventos personalizado, la configuración es similar a la siguiente:

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

En este ejemplo se usa un punto de conexión de storage como destino:

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

En el caso de los puntos de conexión MQTT, también puede usar variables de tema dinámico:

destinationSettings: {
  endpointRef: 'default'
  dataDestination: 'processed/factory/${inputTopic.2}'
}

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>

Para devolver datos al agente MQTT local, use un tema de MQTT estático en la siguiente configuración:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Si tiene un punto de conexión de centro de eventos personalizado, la configuración es similar a la siguiente:

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

En este ejemplo se usa un punto de conexión de storage como destino:

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

En el caso de los puntos de conexión MQTT, también puede usar variables de tema dinámico:

destinationSettings:
  endpointRef: default
  dataDestination: processed/factory/${inputTopic.2}

Use el comando az iot ops dataflow apply para crear o cambiar un data flow.

az iot ops dataflow apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --profile <DataflowProfileName> --name <DataflowName> --config-file <ConfigFilePathAndName>

El --config-file parámetro es la ruta de acceso y el nombre de archivo de un archivo de configuración JSON que contiene las propiedades del recurso.

En este ejemplo, supongamos que se almacena un archivo de configuración denominado data-flow.json con el siguiente contenido en el directorio principal del usuario:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature/#",
            "humidifiers/+/sensor/humidity/#"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "builtInTransformationSettings": {
          "datasets": [],
          "filter": [
            {
              "expression": "$1 * $2 < 100000",
              "inputs": [
                "temperature.Value",
                "\"Tag 10\".Value"
              ],
              "type": "Filter"
            }
          ],
          "map": [
            {
              "inputs": [
                "*"
              ],
              "output": "*",
              "type": "PassThrough"
            },
            {
              "expression": "fToC($1)",
              "inputs": [
                "Temperature.Value"
              ],
              "output": "TemperatureF",
              "type": "Compute"
            },
            {
              "inputs": [
                "@\"Tag 10\".Value"
              ],
              "output": "Humidity",
              "type": "Rename"
            }
          ],
          "serializationFormat": "Json"
        },
        "operationType": "BuiltInTransformation"
      },
      {
        "destinationSettings": {
          "dataDestination": "factory",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Este es un comando de ejemplo para crear o actualizar un data flow mediante el perfil de flujo de datos predeterminado:

az iot ops dataflow apply --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --config-file ~/data-flow.json

Este es otro ejemplo que usa la traducción de temas dinámica para enrutar mensajes de diferentes termostatos a temas específicos del dispositivo:

{
    "mode": "Enabled",
    "operations": [
      {
        "operationType": "Source",
        "sourceSettings": {
          "dataSources": [
            "thermostats/+/sensor/temperature"
          ],
          "endpointRef": "default",
          "serializationFormat": "Json"
        }
      },
      {
        "destinationSettings": {
          "dataDestination": "processed/device/${inputTopic.2}/temperature",
          "endpointRef": "default"
        },
        "operationType": "Destination"
      }
    ]
}

Esta configuración procesa los mensajes de thermostats/device1/sensor/temperature y los envía a processed/device/device1/temperature.

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

// Pointer to the default data flow 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 data flow 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/v1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  # Reference to the default data flow 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

Seleccione el data flow que desea exportar y, a continuación, seleccione Export en la barra de herramientas.

Captura de pantalla de la interfaz de experiencia de operaciones que muestra la opción de exportación para un flujo de datos configurado.

Use el comando az iot ops dataflow show para exportar un data flow.

az iot ops dataflow show --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <DataflowName> --profile <DataflowProfileName> --output json > my-dataflow.json

Este es un comando de ejemplo para exportar un data flow denominado data-flow a un archivo JSON denominado data-flow.json:

az iot ops dataflow show --resource-group myResourceGroup --instance myAioInstance --profile default --name data-flow --output json > data-flow.json

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

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