Creación de un grafo de flujo de datos en operaciones de Azure IoT

Un gráfico de flujo de datos es una canalización de procesamiento compuesta que transforma los datos a medida que se mueven entre orígenes y destinos. Un flujo de datos estándar sigue una secuencia de enriquecimiento, filtro y mapa fija. Un gráfico de flujo de datos permite encadenar transformaciones en cualquier orden, bifurcar en rutas paralelas y agregar datos a lo largo de las ventanas de tiempo.

En este artículo se explica cómo crear un grafo de flujo de datos paso a paso. Para obtener información general sobre los gráficos de flujo de datos y las transformaciones disponibles, consulte Introducción a los gráficos de flujo de datos.

Actualmente, los gráficos de flujo de datos solo admiten puntos de conexión MQTT, Kafka y OpenTelemetry. No soportan otros tipos de endpoints como Data Lake, Microsoft Fabric OneLake, Azure Data Explorer y Local Storage.

Las transformadas utilizan un lenguaje de expresiones para calcular valores, condiciones de prueba y campos de referencia. Las expresiones se refieren a las entradas por posición, no por nombre: la primera entrada en la inputs lista es $1, la segunda es $2, y así sucesivamente. Las funciones integradas, como cToF, convierten y manipulan esos valores.

Para la lista completa de operadores, funciones, tipos de datos y campos de metadatos, consulte la referencia Expressions.

Prerrequisitos

Instancia de Operaciones de IoT de Azure implementada en un clúster de Kubernetes. Para obtener más información, consulte Deploy Operaciones de IoT de Azure.

CLI de Azure está instalado en la máquina de desarrollo. Consulta las extensiones disponibles de CLI de Azure para la versión mínima requerida para usar la extensión azure-iot-ops. Use az --version para comprobar la versión y az upgrade para actualizarla si es necesario. Para más información, consulte Instalación de la CLI de Azure.
Extensión Operaciones de IoT de Azure para el CLI de Azure. Use el siguiente comando para agregar la extensión o actualizarla a la versión más reciente:
```
az extension add --upgrade --name azure-iot-ops
```

Operaciones de IoT de Azure versión 1.2 o posterior.
Un perfil de flujo de datos. Puede usar el perfil predeterminado.
Un punto de conexión de flujo de datos para origen y destino. El punto de conexión de MQTT broker predeterminado permite empezar.

Los CLI de Azure ejemplos de este artículo usan variables de entorno para que puedas establecer cada valor una vez y luego copiar y pegar los comandos as-is. Si usas el entorno Operaciones de IoT de Azure Codespaces del quickstart, estas variables ya están configuradas para ti y puedes saltarte este paso. De lo contrario, configura las siguientes variables de entorno en tu shell antes de ejecutar los comandos.

Los siguientes scripts establecen las variables de entorno más usadas:

Variable del entorno	Descripción
`SUBSCRIPTION_ID`	El ID de la suscripción que contiene tu instancia de Operaciones de IoT de Azure.
`RESOURCE_GROUP`	El nombre del grupo de recursos que contiene tu instancia de Operaciones de IoT de Azure.
`AIO_INSTANCE_NAME`	El nombre de tu instancia de Operaciones de IoT de Azure. Para listar tus instancias, ejecuta `az iot ops list -o table`.
`CLUSTER_NAME`	El nombre del clúster Kubernetes habilitado para Azure Arc que aloja tu instancia.
`LOCATION`	La región de Azure para usar para nuevos recursos, por ejemplo `eastus`.

Bash
PowerShell

SUBSCRIPTION_ID=<subscription-id>
RESOURCE_GROUP=<resource-group-name>
AIO_INSTANCE_NAME=<instance-name>
CLUSTER_NAME=<cluster-name>
LOCATION=<region>

$SUBSCRIPTION_ID = "<subscription-id>"
$RESOURCE_GROUP = "<resource-group-name>"
$AIO_INSTANCE_NAME = "<instance-name>"
$CLUSTER_NAME = "<cluster-name>"
$LOCATION = "<region>"

Solo necesitas establecer las variables que utiliza este artículo. Este artículo podría utilizar variables de entorno adicionales para los nombres de recursos que elijas. El artículo explica cómo situarlos donde se introducen.

Este artículo también utiliza las GRAPH_NAME variables de entorno y PROFILE para el grafo de flujo de datos y los nombres de perfil. Configura cada uno antes de ejecutar los comandos relacionados.

Creación de un gráfico de flujo de datos

Un gráfico de flujo de datos contiene tres tipos de elementos: orígenes que aportan datos, transforman los procesos y destinos que los envían. Conéctelos en el orden en que desea que fluyan los datos.

En la experiencia de operaciones, accede a tu instancia de Operaciones de IoT de Azure.
Seleccione Gráfico de flujo de datos>Crear gráfico de flujo de datos.
Escriba un nombre para el gráfico de flujo de datos y seleccione un perfil de flujo de datos. El perfil predeterminado está seleccionado de forma predeterminada.
Compile la canalización agregando elementos al lienzo:
1. Agregar un origen: seleccione el punto de conexión de origen y configure los temas a los que suscribirse para los mensajes entrantes.
2. Agregar transformaciones: seleccione una o varias transformaciones para procesar los datos. Las transformaciones disponibles incluyen asignación, filtrado, bifurcación, concatenación, ventana y limitación. Para más información sobre cada tipo de transformación, consulte Introducción a los gráficos de flujo de datos.
3. Agregar un destino: seleccione el punto de conexión de destino y configure el tema o la ruta de acceso para enviar datos procesados.
Conecte los elementos en el orden en que desea que fluyan los datos.
Seleccione Guardar para implementar el gráfico de flujo de datos.

La CLI de Azure utiliza un grafo de flujo de datos a partir de un único archivo de configuración JSON que contiene todos los nodos y conexiones. Use az iot ops dataflowgraph apply para crear o reemplazar el grafo. El siguiente ejemplo lee datos de temperatura, los convierte a Fahrenheit y los envía a un tema de destino.

Cree un graph.json archivo con las propiedades del grafo de flujo de datos. En el graph.json archivo, almacena las reglas de cada transformación en el value campo como una cadena JSON escapada. Para obtener la forma legible de las reglas de cada transformación, consulte el procedimiento para ese tipo de transformación.

{
  "mode": "Enabled",
  "nodes": [
    {
      "nodeType": "Source",
      "name": "sensors",
      "sourceSettings": {
        "endpointRef": "default",
        "dataSources": [
          "telemetry/temperature"
        ]
      }
    },
    {
      "nodeType": "Graph",
      "name": "convert",
      "graphSettings": {
        "registryEndpointRef": "default",
        "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
        "configuration": [
          {
            "key": "rules",
            "value": "{\"map\":[{\"inputs\":[\"*\"],\"output\":\"*\"},{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
          }
        ]
      }
    },
    {
      "nodeType": "Destination",
      "name": "output",
      "destinationSettings": {
        "endpointRef": "default",
        "dataDestination": "telemetry/converted"
      }
    }
  ],
  "nodeConnections": [
    {
      "from": {
        "name": "sensors"
      },
      "to": {
        "name": "convert"
      }
    },
    {
      "from": {
        "name": "convert"
      },
      "to": {
        "name": "output"
      }
    }
  ]
}

Aplique el archivo de configuración.

az iot ops dataflowgraph apply \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP \
  --config-file graph.json

El gráfico utiliza el perfil de default flujo de datos. Para usar otro perfil, agregue --profile <PROFILE_NAME>.

Cree un archivo de Bicep .bicep con la siguiente estructura. En este ejemplo se crea un gráfico de flujo de datos que lee los datos de temperatura, lo convierte en Fahrenheit y lo envía a un tema de destino.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'

resource aioInstance 'Microsoft.IoTOperations/instances@2026-03-01' existing = {
  name: aioInstanceName
}

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2026-03-01' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: defaultDataflowProfile
  name: 'temperature-processing'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    profileRef: 'default'
    mode: 'Enabled'
    nodes: [
      {
        nodeType: 'Source'
        name: 'sensors'
        sourceSettings: {
          endpointRef: 'default'
          dataSources: [
            'telemetry/temperature'
          ]
        }
      }
      {
        nodeType: 'Graph'
        name: 'convert'
        graphSettings: {
          registryEndpointRef: 'default'
          artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
          configuration: [
            {
              key: 'rules'
              value: '{"map":[{"inputs":["*"],"output":"*"},{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
            }
          ]
        }
      }
      {
        nodeType: 'Destination'
        name: 'output'
        destinationSettings: {
          endpointRef: 'default'
          dataDestination: 'telemetry/converted'
        }
      }
    ]
    nodeConnections: [
      {
        from: { name: 'sensors' }
        to: { name: 'convert' }
      }
      {
        from: { name: 'convert' }
        to: { name: 'output' }
      }
    ]
  }
}

Implemente el archivo de Bicep:

az deployment group create --resource-group $RESOURCE_GROUP --template-file main.bicep

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

Cree un archivo de manifiesto .yaml de Kubernetes con la siguiente estructura. En este ejemplo se crea un gráfico de flujo de datos que lee los datos de temperatura, lo convierte en Fahrenheit y lo envía a un tema de destino.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: temperature-processing
  namespace: azure-iot-operations
spec:
  profileRef: default
  nodes:
    - nodeType: Source
      name: sensors
      sourceSettings:
        endpointRef: default
        dataSources:
          - telemetry/temperature

    - nodeType: Graph
      name: convert
      graphSettings:
        registryEndpointRef: default
        artifact: azureiotoperations/graph-dataflow-map:1.0.0
        configuration:
          - key: rules
            value: |
              {
                "map": [
                  {
                    "inputs": ["*"],
                    "output": "*"
                  },
                  {
                    "inputs": ["temperature"],
                    "output": "temperature_f",
                    "expression": "cToF($1)"
                  }
                ]
              }

    - nodeType: Destination
      name: output
      destinationSettings:
        endpointRef: default
        dataDestination: telemetry/converted

  nodeConnections:
    - from:
        name: sensors
      to:
        name: convert
    - from:
        name: convert
      to:
        name: output

Aplique el manifiesto:

kubectl apply -f main.yaml

Configuración del origen

El origen define dónde entran los datos en la canalización. Especifique una referencia de punto de conexión y uno o varios temas.

En el editor de gráficos de flujo de datos, seleccione el elemento de origen y configure:

Configuración	Descripción
Punto final	Punto de conexión de flujo de datos a utilizar. Seleccione el valor predeterminado para el agente MQTT local.
Temas	Uno o varios temas a los que suscribirse para los mensajes entrantes.

La CLI aplica el grafo completo de una sola vez, por lo que configure el origen como un nodo Source en el archivo de configuración graph.json y, a continuación, ejecute az iot ops dataflowgraph apply:

{
  "nodeType": "Source",
  "name": "sensors",
  "sourceSettings": {
    "endpointRef": "default",
    "dataSources": [
      "telemetry/temperature",
      "telemetry/humidity"
    ]
  }
}

{
  nodeType: 'Source'
  name: 'sensors'
  sourceSettings: {
    endpointRef: 'default'
    dataSources: [
      'telemetry/temperature'
      'telemetry/humidity'
    ]
  }
}

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

- nodeType: Source
  name: sensors
  sourceSettings:
    endpointRef: default
    dataSources:
      - telemetry/temperature
      - telemetry/humidity

Añadir transformaciones

Transforma los datos de procesamiento entre el origen y el destino. Cada transformación hace referencia a un artefacto incorporado y tiene reglas de configuración.

Las transformaciones integradas disponibles son:

Transformación	Artefacto	Descripción
Map	`azureiotoperations/graph-dataflow-map:1.0.0`	Cambiar el nombre, reestructurar, calcular y copiar campos.
Filter	`azureiotoperations/graph-dataflow-filter:1.0.0`	Descartar los mensajes que cumplan una condición.
Rama	`azureiotoperations/graph-dataflow-branch:1.0.0`	Dirija cada mensaje a una ruta `true` o `false` en función de una condición.
Concatenate	`azureiotoperations/graph-dataflow-concatenate:1.0.0`	Combinar dos o más rutas en una sola.
Ventana	`azureiotoperations/graph-dataflow-window:1.0.0`	Recopile mensajes a lo largo de un intervalo de tiempo y, a continuación, agregue.
Throttle	`azureiotoperations/graph-dataflow-throttle:1.0.0`	Limita la tasa de mensajes por patrón de tema MQTT.

Para obtener más información sobre cómo enriquecer mensajes con datos externos, consulte Enriquecimiento con datos externos.

En el editor de gráficos de flujo de datos, seleccione Agregar transformación y elija el tipo de transformación. Configure las reglas en el editor visual.

Cada transformación es un nodo con nodeType establecido en Graph en el archivo de configuración graph.json. Las reglas de la transformación son un objeto JSON, como este mapa que convierte la temperatura en Fahrenheit:

{
  "map": [
    {
      "inputs": ["temperature"],
      "output": "temperature_f",
      "expression": "cToF($1)"
    }
  ]
}

La configuration propiedad toma estas reglas como una cadena, así que salta el JSON de reglas y colócala en el value campo. Aplique el gráfico completo con az iot ops dataflowgraph apply:

{
  "nodeType": "Graph",
  "name": "convert",
  "graphSettings": {
    "registryEndpointRef": "default",
    "artifact": "azureiotoperations/graph-dataflow-map:1.0.0",
    "configuration": [
      {
        "key": "rules",
        "value": "{\"map\":[{\"inputs\":[\"temperature\"],\"output\":\"temperature_f\",\"expression\":\"cToF($1)\"}]}"
      }
    ]
  }
}

Tip

Para generar la cadena escapada, guarda las reglas en un archivo como rules.json, ejecuta jq -c . rules.json, y pega la salida de una sola línea en el value campo.

Cada transformación es un nodo con nodeType: 'Graph'. La configuration propiedad pasa reglas como una cadena JSON:

{
  nodeType: 'Graph'
  name: 'convert'
  graphSettings: {
    registryEndpointRef: 'default'
    artifact: 'azureiotoperations/graph-dataflow-map:1.0.0'
    configuration: [
      {
        key: 'rules'
        value: '{"map":[{"inputs":["temperature"],"output":"temperature_f","expression":"cToF($1)"}]}'
      }
    ]
  }
}

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

Cada transformación es un nodo con nodeType: Graph. La configuration propiedad pasa reglas como una cadena JSON:

- nodeType: Graph
  name: convert
  graphSettings:
    registryEndpointRef: default
    artifact: azureiotoperations/graph-dataflow-map:1.0.0
    configuration:
      - key: rules
        value: |
          {
            "map": [
              {
                "inputs": ["temperature"],
                "output": "temperature_f",
                "expression": "cToF($1)"
              }
            ]
          }

Encadenar varias transformaciones

Encadena cualquier cantidad de transformaciones. Conéctelos en la nodeConnections sección en el orden en que desea que fluyan los datos:

Arrastre las conexiones entre transformaciones en el lienzo para definir el orden de procesamiento.

Defina el orden de procesamiento en la sección nodeConnections de su archivo de configuración graph.json:

"nodeConnections": [
  {
    "from": {
      "name": "sensors"
    },
    "to": {
      "name": "remove-bad-data"
    }
  },
  {
    "from": {
      "name": "remove-bad-data"
    },
    "to": {
      "name": "convert"
    }
  },
  {
    "from": {
      "name": "convert"
    },
    "to": {
      "name": "output"
    }
  }
]

nodeConnections: [
  { from: { name: 'sensors' }, to: { name: 'remove-bad-data' } }
  { from: { name: 'remove-bad-data' }, to: { name: 'convert' } }
  { from: { name: 'convert' }, to: { name: 'output' } }
]

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

nodeConnections:
  - from: { name: sensors }
    to: { name: remove-bad-data }
  - from: { name: remove-bad-data }
    to: { name: convert }
  - from: { name: convert }
    to: { name: output }

Configuración del destino

El destino define hacia dónde van los datos procesados. Especifique una referencia de punto de conexión y un tema o ruta de acceso.

Seleccione el elemento de destino y configure:

Configuración	Descripción
Punto final	Punto de conexión de flujo de datos al que se van a enviar datos.
Topic	Tema o ruta de acceso a la que publicar los datos procesados.

Configure el destino como un nodo Destination en su archivo de configuración graph.json, y luego aplique el grafo completo con az iot ops dataflowgraph apply:

{
  "nodeType": "Destination",
  "name": "output",
  "destinationSettings": {
    "endpointRef": "default",
    "dataDestination": "telemetry/processed"
  }
}

{
  nodeType: 'Destination'
  name: 'output'
  destinationSettings: {
    endpointRef: 'default'
    dataDestination: 'telemetry/processed'
  }
}

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

- nodeType: Destination
  name: output
  destinationSettings:
    endpointRef: default
    dataDestination: telemetry/processed

Para el enrutamiento dinámico de temas en función del contenido del mensaje, consulte Enrutamiento de mensajes a distintos temas.

Comprobación de que el gráfico de flujo de datos funciona

Después de implementar un gráfico de flujo de datos, compruebe que se está ejecutando:

En la experiencia de operaciones, selecciona tu gráfico de flujo de datos para ver su estado. Un gráfico correcto muestra un estado En ejecución .

Use az iot ops dataflowgraph show para ver los detalles de un grafo:

az iot ops dataflowgraph show \
  --name temperature-processing \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Para enumerar todos los gráficos de flujo de datos asociados a un perfil, use az iot ops dataflowgraph list:

az iot ops dataflowgraph list \
  --instance $AIO_INSTANCE_NAME \
  --resource-group $RESOURCE_GROUP

Compruebe el estado del DataflowGraph recurso:

az resource show --resource-group $RESOURCE_GROUP --resource-type Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs --name $GRAPH_NAME --parent instances/$AIO_INSTANCE_NAME/dataflowProfiles/$PROFILE

Importante

El uso de manifiestos de implementación de Kubernetes no se admite en entornos de producción y solo se debe usar para la depuración y las pruebas.

kubectl get dataflowgraph temperature-processing -n azure-iot-operations

Compruebe si hay errores en los registros del pod:

kubectl logs -l app=dataflow -n azure-iot-operations --tail=50

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-07-31

Creación de un grafo de flujo de datos en operaciones de Azure IoT

Prerrequisitos

Creación de un gráfico de flujo de datos

Configuración del origen

Añadir transformaciones

Encadenar varias transformaciones

Configuración del destino

Comprobación de que el gráfico de flujo de datos funciona

Contenido relacionado

Comentarios

Recursos adicionales