Criar fluxos de dados nas Operações de IoT do Azure

1. Para criar um fluxo de dados na experiência de operações, selecione Fluxo de dados>Criar fluxo de dados.

2. Selecione o nome do espaço reservado new-data-flow para definir as propriedades do fluxo de dados. Insira o nome do fluxo de dados e escolha o perfil de fluxo de dados a ser usado. O perfil de fluxo de dados padrão é selecionado por padrão. Para obter mais informações sobre perfis de fluxo de dados, consulte Configurar o perfil de fluxo de dados.

   

   Importante

   Você só pode escolher o perfil de fluxo de dados ao criar um fluxo de dados. Você não pode alterar o perfil de fluxo de dados depois que o fluxo de dados é criado. Se você quiser alterar o perfil de fluxo de dados de um fluxo de dados existente, exclua o fluxo de dados original e crie um novo com o novo perfil de fluxo de dados.

3. Configure o ponto de extremidade de origem, transformação e destino para o fluxo de dados selecionando os itens no diagrama de fluxo de dados.

   

Use o comando az iot ops dataflow apply para criar ou alterar um fluxo de dados.

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

O --config-file parâmetro é o caminho e o nome do arquivo de um arquivo de configuração JSON que contém as propriedades do recurso.

Neste exemplo, suponha um arquivo de configuração nomeado data-flow.json com o seguinte conteúdo armazenado no diretório base do usuário:

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

Aqui está um comando de exemplo para criar ou atualizar um fluxo de dados usando o perfil de fluxo de dados padrão:

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

Crie um arquivo .bicep do Bicep para começar a criar um fluxo de dados. Este exemplo mostra a estrutura do fluxo de dados que contém as configurações de origem, transformação e 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
        }
      }
    ]
  }
}

Crie um arquivo .yaml de manifesto do Kubernetes para começar a criar um fluxo de dados. Este exemplo mostra a estrutura do fluxo de dados que contém as configurações de origem, transformação e 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. Em Detalhes da fonte, selecione Agente de mensagens.

   

2. Insira as seguintes configurações para a fonte do agente de mensagens:

   Configuração	Descrição
   Ponto de extremidade de fluxo de dados	Selecione padrão para usar o ponto de extremidade do agente de mensagens MQTT padrão.
   Tópico	O filtro de tópico a ser assinado para as mensagens de entrada. Use Tópicos>Adicionar linha para adicionar vários tópicos. Para obter mais informações sobre tópicos, confira Configurar tópicos MQTT ou Kafka.
   Esquema de mensagem	O esquema a ser usado para desserializar as mensagens de entrada. Confira Especificar o esquema para desserializar dados.

3. Escolha Aplicar.

Veja um exemplo de configuração de ponto de extremidade da fonte para o ponto de extremidade do agente MQTT padrão:

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

O endpoint do broker de mensagens está configurado no arquivo Bicep. Por exemplo, o ponto de extremidade a seguir é uma fonte para o fluxo de dados.

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

Por exemplo, para configurar uma fonte usando um ponto de extremidade de agente de mensagens e dois filtros de tópicos, use a seguinte configuração:

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

Você pode usar um ativo como a fonte do fluxo de dados. O uso de um ativo como fonte só está disponível na experiência de operações.

1. Em Detalhes da fonte, selecione Ativo.

2. Selecione o ativo que você deseja usar como ponto de extremidade da fonte.

3. Selecione Continuar.

   Uma lista de pontos de dados para o ativo selecionado será exibida.

   

4. Selecione Aplicar para usar o ativo como o ponto de extremidade da fonte.

A configuração de um ativo como fonte só está disponível na experiência de operações.

A configuração de um ativo como fonte só está disponível na experiência de operações.

A configuração de um ativo como fonte só está disponível na experiência de operações.

1. Em Detalhes da fonte, selecione Agente de mensagens.

   

2. Insira as seguintes configurações para a fonte do agente de mensagens:

   Configuração	Descrição
   Ponto de extremidade de fluxo de dados	Use o botão Selecionar novamente para selecionar um ponto de extremidade de fluxo de dados MQTT ou Kafka personalizado. Para obter mais informações, confira Configurar pontos de extremidade de fluxo de dados MQTT ou Configurar pontos de extremidade de fluxo de dados dos Hubs de Eventos do Azure e do Kafka.
   Tópico	O filtro de tópico a ser assinado para as mensagens de entrada. Use Tópicos>Adicionar linha para adicionar vários tópicos. Para obter mais informações sobre tópicos, confira Configurar tópicos MQTT ou Kafka.
   Esquema de mensagem	O esquema a ser usado para desserializar as mensagens de entrada. Confira Especificar o esquema para desserializar dados.

3. Escolha Aplicar.

Substitua os valores de espaço reservado por seu nome de ponto de extremidade personalizado e tópicos.

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

Substitua os valores de espaço reservado por seu nome de ponto de extremidade personalizado e tópicos.

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

Substitua os valores de espaço reservado por seu nome de ponto de extremidade personalizado e tópicos.

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

No fluxo de dados de experiência de operações Detalhes da fonte, selecione Agente de mensagens e, em seguida, use o campo Tópicos para especificar os filtros de tópico MQTT para assinar mensagens recebidas. Você pode adicionar vários tópicos MQTT selecionando Adicionar linha e inserindo um novo tópico.

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

Exemplo com vários filtros de tópico MQTT com curingas:

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

Aqui, o curinga + é usado para selecionar todos os dispositivos nos tópicos thermostats e humidifiers. O curinga # é usado para selecionar todas as mensagens do sensor em todos os subtópicos dos tópicos temperature e humidity.

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

Exemplo com vários filtros de tópico MQTT com curingas:

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

Aqui, o curinga + é usado para selecionar todos os dispositivos nos tópicos thermostats e humidifiers. O curinga # é usado para selecionar todas as mensagens do sensor em todos os subtópicos dos tópicos temperature e humidity.

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

Exemplo com vários filtros de tópico com curingas:

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

Aqui, o curinga + é usado para selecionar todos os dispositivos nos tópicos thermostats e humidifiers. O curinga # é usado para selecionar todas as mensagens em todos os subtópicos dos tópicos temperature e humidity.

No fluxo de dados da experiência de operações Detalhes da fonte, selecione Agente de mensagens e use o campo Tópico para especificar o grupo de assinatura compartilhado e o 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>

No fluxo de dados da experiência de operações Detalhes da fonte, selecione Agente de mensagens e use o campo Tópico para especificar o filtro de tópico do Kafka para assinar mensagens de entrada.

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

No fluxo de dados da experiência de operações Detalhes da fonte, selecione Agente de mensagens e use o campo Esquema de mensagem para especificar o esquema. Você pode usar o botão Carregar para carregar um arquivo de esquema primeiro. Para saber mais, confira Entender os esquemas de mensagens.

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

Depois de usar o registro de esquema para armazenar o esquema, você poderá referenciá-lo na configuração do fluxo de dados.

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

Depois de usar o registro de esquema para armazenar o esquema, você poderá referenciá-lo na configuração do fluxo de dados.

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

Ao criar ou editar um fluxo de dados, selecione Editar e selecione Sim ao lado de Solicitar persistência de dados.

Adicione a propriedade requestDiskPersistence ao arquivo de configuração do fluxo de dados:

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

Adicione a propriedade requestDiskPersistence ao recurso de fluxo de dados. A versão da API é 2025-10-01 ou 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
    ]
  }
}

Adicione a propriedade requestDiskPersistence à especificação de fluxo de dados. A versão da API é connectivity.iotoperations.azure.com/v1beta1 ou 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

Na experiência de operações, selecione Fluxo de dados>Adicionar transformação (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

Atualmente, não há suporte para o estágio Enriquecer na experiência de operações.

Para enriquecer os dados, você pode usar a builtInTransformationSettings propriedade na configuração de fluxo de dados. A datasets propriedade é usada para especificar os conjuntos de dados a serem usados para enriquecimento.

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

Esse exemplo mostra como você pode usar o campo deviceId nos dados da fonte para corresponder ao campo asset no conjunto de dados:

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

Por exemplo, você pode usar o campo deviceId nos dados da fonte para corresponder ao campo asset no conjunto de dados:

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

1. Em Transformar (opcional), selecione Filtro>Adicionar.

   

2. Insira as configurações necessárias.

   Configuração	Descrição
   Condição de filtro	A condição para filtrar os dados com base em um campo nos dados da fonte.
   Descrição	Forneça uma descrição para a condição de filtro.

   No campo de condição de filtro, insira @ ou selecione Ctrl + Espaço para escolher pontos de dados em uma lista suspensa.

   Você pode inserir propriedades de metadados MQTT usando o formato @$metadata.user_properties.<property> ou @$metadata.topic. Você também pode inserir cabeçalhos $metadata usando o formato @$metadata.<header>. A sintaxe $metadata só é necessária para propriedades MQTT que fazem parte do cabeçalho da mensagem. Para mais informações, confira referências de campo.

   A condição pode usar os campos nos dados da fonte. Por exemplo, você pode usar uma condição de filtro como @temperature > 20 para filtrar dados menores ou iguais a 20 com base no campo de temperatura.

3. Escolha Aplicar.

Por exemplo, você pode usar o campo temperature nos dados da fonte para filtrar os dados:

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

Por exemplo, você pode usar o campo temperature nos dados da fonte para filtrar os dados:

builtInTransformationSettings: {
  filter: [
    {
      inputs: [
        'temperature ? $last'
      ]
      expression: '$1 > 20'
    }
  ]
}

Se o campo temperature for maior que 20, os dados serão passados para o próximo estágio. Se o campo temperature for menor ou igual a 20, os dados serão filtrados.

Por exemplo, você pode usar o campo temperature nos dados da fonte para filtrar os dados:

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

Se o campo temperature for maior que 20, os dados serão passados para o próximo estágio. Se o campo temperature for menor ou igual a 20, os dados serão filtrados.

1. Em Transformar (opcional), selecione Nova propriedade>Adicionar.

   

2. Insira as configurações necessárias.

   Configuração	Descrição
   Chave da propriedade	Insira a chave para a nova propriedade.
   Valor da propriedade	Insira o valor da nova propriedade.
   Descrição	Forneça uma descrição para a nova propriedade.

3. Escolha Aplicar.

Por exemplo, você pode usar o campo temperature nos dados da fonte para converter a temperatura em Celsius e armazená-la no campo temperatureCelsius. Você também pode enriquecer os dados da fonte com o campo location do conjunto de dados de contextualização:

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

É possível acessar as propriedades de metadados do MQTT usando o formato $metadata.user_properties.<property> ou $metadata.topic. Você também pode inserir cabeçalhos $metadata usando o formato $metadata.<header>. Para mais informações, confira referências de campo.

Por exemplo, você pode usar o campo temperature nos dados da fonte para converter a temperatura em Celsius e armazená-la no campo temperatureCelsius. Você também pode enriquecer os dados da fonte com o campo location do conjunto de dados de contextualização:

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

É possível acessar as propriedades de metadados do MQTT usando o formato $metadata.user_properties.<property> ou $metadata.topic. Você também pode inserir cabeçalhos $metadata usando o formato $metadata.<header>. Para mais informações, confira referências de campo.

Por exemplo, você pode usar o campo temperature nos dados da fonte para converter a temperatura em Celsius e armazená-la no campo temperatureCelsius. Você também pode enriquecer os dados da fonte com o campo location do conjunto de dados de contextualização:

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

1. Em Transformar (opcional),selecione Remover.

2. Selecione o ponto de dados a ser removido do esquema de saída.

   

3. Escolha Aplicar.

Para remover um ponto de dados do esquema de saída, você pode usar a builtInTransformationSettings propriedade na configuração de fluxo de dados. A map propriedade é usada para especificar os pontos de dados a serem removidos.

{
  "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 experiência de operações, especifique o esquema e o formato de serialização nos detalhes do ponto de extremidade do fluxo de dados. Os pontos de extremidade que dão suporte a formatos de serialização são o OneLake do Microsoft Fabric, o Azure Data Lake Storage Gen2, o Azure Data Explorer e o armazenamento local. Por exemplo, para serializar os dados no formato Delta, você precisa carregar um esquema para o registro de esquema e referenciá-lo na configuração do ponto final de destino do fluxo de dados.

Depois de carregar um esquema no registro de esquema, você poderá referenciá-lo na configuração do fluxo de dados.

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

Depois de carregar um esquema no registro de esquema, você poderá referenciá-lo na configuração do fluxo de dados.

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

Depois de carregar um esquema no registro de esquema, você poderá referenciá-lo na configuração do fluxo de dados.

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

1. Selecione o ponto de extremidade de fluxo de dados a ser usado como destino.

   

   Os pontos de extremidade de armazenamento exigem um esquema para serialização. Se você escolher um ponto de extremidade de destino do OneLake do Microsoft Fabric, do Azure Data Lake Storage, do Azure Data Explorer ou do Local Storage, deverá especificar uma referência de esquema. Por exemplo, para serializar os dados em um ponto de extremidade do Microsoft Fabric no formato Delta, você precisa carregar um esquema no registro de esquemas e referenciá-lo na configuração do ponto de extremidade de destino do fluxo de dados.

   

2. Selecione Continuar para configurar o destino.

3. Insira as configurações necessárias para o destino, incluindo o tópico ou a tabela para o qual enviar os dados. Confira Configurar o destino dos dados (tópico, contêiner ou tabela) para obter mais informações.

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

Ao usar a experiência de operações, o campo de destino dos dados é interpretado automaticamente com base no tipo de ponto de extremidade. Por exemplo, se o ponto de extremidade de fluxo de dados for um ponto de extremidade de armazenamento, a página de detalhes do destino solicitará que você insira o nome do contêiner. Se o ponto de extremidade de fluxo de dados for um ponto de extremidade MQTT, a página de detalhes do destino solicitará que você insira o tópico e assim por diante.

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

Por exemplo, para enviar dados de volta ao agente MQTT local em um tópico MQTT estático, use a seguinte configuração:

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

Ou, se você tiver um ponto de extremidade de hub de eventos personalizado, a configuração será semelhante:

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

Para pontos de extremidade MQTT, você também pode usar variáveis de tópico dinâmico. Por exemplo, para rotear mensagens de factory/1/data para processed/factory/1:

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

A sintaxe é a mesma para todos os pontos de extremidade de fluxo de dados:

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

Por exemplo, para enviar dados de volta ao agente MQTT local em um tópico MQTT estático, use a seguinte configuração:

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

Ou, se você tiver um ponto de extremidade de hub de eventos personalizado, a configuração será semelhante:

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

Outro exemplo usando um ponto de extremidade de armazenamento como destino:

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

Para pontos de extremidade MQTT, você também pode usar variáveis de tópico dinâmico:

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

A sintaxe é a mesma para todos os pontos de extremidade de fluxo de dados:

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

Por exemplo, para enviar dados de volta ao agente MQTT local em um tópico MQTT estático, use a seguinte configuração:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Ou, se você tiver um ponto de extremidade de hub de eventos personalizado, a configuração será semelhante:

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

Outro exemplo usando um ponto de extremidade de armazenamento como destino:

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

Para pontos de extremidade MQTT, você também pode usar variáveis de tópico dinâmico:

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

Use o comando az iot ops dataflow apply para criar ou alterar um fluxo de dados.

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

O --config-file parâmetro é o caminho e o nome do arquivo de um arquivo de configuração JSON que contém as propriedades do recurso.

Neste exemplo, suponha um arquivo de configuração nomeado data-flow.json com o seguinte conteúdo armazenado no diretório base do usuário:

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

Aqui está um comando de exemplo para criar ou atualizar um fluxo de dados usando o perfil de fluxo de dados padrão:

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

Aqui está outro exemplo usando a tradução de tópico dinâmico para rotear mensagens de termostatos diferentes para tópicos específicos do 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"
      }
    ]
}

Essa configuração processa mensagens de thermostats/device1/sensor/temperature e as envia para 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

Selecione o fluxo de dados que deseja exportar e selecione Exportar na barra de ferramentas.

Use o comando az iot ops dataflow show para exportar um fluxo de dados.

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

Aqui está um comando de exemplo para exportar um fluxo de dados nomeado data-flow para um arquivo JSON chamado data-flow.json:

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

O Bicep é uma infraestrutura como código e nenhuma exportação é necessária. Use o Arquivo Bicep para criar um fluxo de dados para configurar e definir rapidamente fluxos de dados.

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