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

Importante

Esta página inclui instruções para gerir componentes do Azure IoT Operations usando manifestos de implementação do Kubernetes, que estão em pré-visualização. Esse recurso é fornecido com várias limitações e não deve ser usado para cargas de trabalho de produção.

Consulte os Termos de Utilização Suplementares das Pré-visualizações do Microsoft Azure para obter os termos legais que se aplicam às funcionalidades do Azure que estão em versão beta, pré-visualização ou ainda não disponibilizadas para disponibilidade geral.

Um fluxo de dados é o caminho que os dados percorrem da origem para o destino com transformações opcionais. Você pode configurar o fluxo de dados criando um recurso personalizado de fluxo de dados ou usando a UI da experiência de operações na Web. Um fluxo de dados é composto por três partes: a origem, a transformação e o destino.

Para definir a origem e o destino, você precisa configurar os pontos de extremidade de fluxo de dados. A transformação é opcional e pode incluir operações como enriquecer os dados, filtrar os dados e mapeá-los para outro campo.

Importante

Cada fluxo de dados deve ter o ponto de extremidade padrão do agente MQTT local do Azure IoT Operations como origem ou destino.

Você pode usar a experiência de operações no Azure IoT Operations para criar um fluxo de dados. A experiência de operações fornece uma interface visual para configurar o fluxo de dados. Você também pode usar o Bicep para criar um fluxo de dados usando um arquivo Bicep ou usar o Kubernetes para criar um fluxo de dados usando um arquivo YAML.

Continue lendo para saber como configurar a origem, a transformação e o destino.

Pré-requisitos

Você pode implantar fluxos de dados logo que tiver uma instância das Operações do Azure IoT usando o perfil de fluxo de dados padrão e o ponto de extremidade. No entanto, convém configurar perfis de fluxo de dados e pontos de extremidade para personalizar o fluxo de dados.

Perfil de fluxo de dados

Se você não precisar de configurações de dimensionamento diferentes para seus fluxos de dados, use o perfil de fluxo de dados padrão fornecido pelas Operações IoT do Azure. Você deve evitar associar muitos fluxos de dados a um único perfil de fluxo de dados. Se você tiver um grande número de fluxos de dados, distribua-os entre vários perfis de fluxo de dados para reduzir o risco de exceder o limite de tamanho de configuração do perfil de fluxo de dados de 70.

Para saber como configurar um novo perfil de fluxo de dados, consulte Configurar perfis de fluxo de dados.

Pontos finais de fluxo de dados

Os pontos de extremidade de fluxo de dados são necessários para configurar a origem e o destino do fluxo de dados. Para começares rapidamente, podes usar o endpoint padrão de fluxo de dados para o broker MQTT local. Você também pode criar outros tipos de pontos de extremidade de fluxo de dados, como Kafka, Hubs de Eventos, OpenTelemetry ou Armazenamento Azure Data Lake. Para saber como configurar cada tipo de ponto de extremidade de fluxo de dados, consulte Configurar pontos de extremidade de fluxo de dados.

Introdução

Depois de ter os pré-requisitos, você pode começar a criar um fluxo de dados.

Experiência em operações

1. Para criar um fluxo de dados na experiência de operações, selecione Fluxo de> dadosCriar 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 perfil de fluxo de dados.

   

   Importante

   Você só pode escolher o perfil de fluxo de dados ao criar um fluxo de dados. Não é possível alterar o perfil de fluxo de dados depois que o fluxo de dados é criado. Se desejar 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.

   

CLI do Azure

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 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 inicial 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 exemplo de comando 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

Bíceps

Crie um arquivo Bicep .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
        }
      }
    ]
  }
}

Kubernetes (visualização)

Crie um arquivo de manifesto .yaml 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

Analise as seções a seguir para saber como configurar os tipos de operação do fluxo de dados.

Fonte

Para configurar uma fonte para o fluxo de dados, especifique a referência de ponto de extremidade e uma lista de fontes de dados para o ponto de extremidade. Escolha uma das seguintes opções como a fonte para o fluxo de dados.

Se o ponto de extremidade padrão não for usado como origem, ele deverá ser usado como destino. Para saber mais sobre como usar o ponto de extremidade do corretor MQTT local, consulte Os fluxos de dados devem usar o ponto de extremidade do corretor MQTT local.

Opção 1: Usar o ponto de extremidade padrão do agente de mensagens como origem

Experiência em operações

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

   

2. Insira as seguintes configurações para a origem do sistema de intermediação de mensagens:

   Configurações	Descrição
   Ponto final do fluxo de dados	Selecione padrão para usar o endpoint padrão do broker de mensagens MQTT.
   Tópico	O filtro de tópico para subscrição para mensagens recebidas. Use Tópico(s)>Adicionar linha para adicionar vários tópicos. Para obter mais informações sobre tópicos, consulte Configurar tópicos MQTT ou Kafka.
   Esquema de mensagem	O esquema a ser usado para desserializar as mensagens de entrada. Consulte Especificar esquema para desserializar dados.

3. Selecione Aplicar.

CLI do Azure

Aqui está um exemplo de configuração de ponto de extremidade de origem para o ponto de extremidade padrão do broker MQTT:

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

Bíceps

O ponto de extremidade do broker de mensagens é configurado no ficheiro Bicep. Por exemplo, o seguinte endpoint é uma fonte de fluxo de dados.

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

Kubernetes (visualização)

Por exemplo, para configurar uma origem usando um ponto de extremidade do corretor de mensagens e dois filtros de tópicos, utilize a seguinte configuração:

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

Como dataSources permite especificar tópicos MQTT ou Kafka sem modificar a configuração do ponto de extremidade, você pode reutilizar o ponto de extremidade para vários fluxos de dados, mesmo que os tópicos sejam diferentes. Para saber mais, consulte Configurar fontes de dados.

Opção 2: Usar ativo como fonte

Experiência em operações

Você pode usar um recurso como fonte para o 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 deseja usar como o ponto de extremidade de origem.

3. Selecione Continuar.

   Uma lista de pontos de dados para o ativo selecionado é exibida.

   

4. Selecione Aplicar para usar o recurso como o ponto de extremidade de origem.

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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

Ao usar um ativo como origem, a definição de ativo é usada para inferir o esquema para o fluxo de dados. A definição de ativo inclui o esquema para os pontos de dados do ativo. Para saber mais, consulte Gerenciar configurações de ativos remotamente.

Uma vez configurados, os dados do ativo alcançam o fluxo de dados através do broker MQTT local. Assim, ao usar um ativo como fonte, o fluxo de dados usa o ponto de extremidade padrão do broker MQTT local como a fonte de facto.

Opção 3: Usar um endpoint de dados personalizado MQTT ou Kafka como origem

Se criou um endpoint de fluxo de dados MQTT ou Kafka personalizado (por exemplo, para utilizar com o Event Grid ou Event Hubs), poderá utilizá-lo como fonte para o fluxo de dados. Lembre-se de que pontos de extremidade de armazenamento, como o Data Lake ou o Fabric OneLake, não podem ser utilizados como fontes.

Experiência em operações

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

   

2. Insira as seguintes configurações para a origem do sistema de intermediação de mensagens:

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

3. Selecione Aplicar.

CLI do Azure

Substitua valores de espaço reservado pelo nome e tópicos personalizados do ponto de extremidade.

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

Bíceps

Substitua valores de espaço reservado pelo nome e tópicos personalizados do ponto de extremidade.

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

Kubernetes (visualização)

Substitua valores de espaço reservado pelo nome e tópicos personalizados do ponto de extremidade.

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

Configurar fontes de dados (tópicos MQTT ou Kafka)

Você pode especificar vários tópicos MQTT ou Kafka em uma fonte sem precisar modificar a configuração do ponto de extremidade do fluxo de dados. Essa flexibilidade significa que o mesmo ponto de extremidade pode ser reutilizado em vários fluxos de dados, mesmo que os tópicos variem. Para obter mais informações, consulte Reutilizar endereços de fluxo de dados.

Tópicos MQTT

Quando a origem é um ponto de extremidade MQTT (incluído Event Grid), pode-se usar o filtro de tópico MQTT para subscrever mensagens de entrada. O filtro de tópicos pode incluir curingas para assinar vários tópicos. Por exemplo, thermostats/+/sensor/temperature/# subscreve todas as mensagens do sensor de temperatura dos termostatos. Para configurar os filtros de tópico MQTT:

Experiência em operações

Nos detalhes da fonte do fluxo de dados de operações, selecione Agente de mensagens e use o campo Tópico(s) para especificar os filtros de tópico MQTT para subscrever mensagens de entrada. Você pode adicionar vários tópicos MQTT selecionando Adicionar linha e inserindo um novo tópico.

CLI do Azure

{
    "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 MQTT para tópicos com curingas:

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

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

Bíceps

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 MQTT para tópicos com curingas:

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

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

Kubernetes (visualização)

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 sob os tópicos thermostats e humidifiers. O curinga # é usado para selecionar todas as mensagens em todos os subtópicos dos tópicos temperature e humidity.

Subscrições partilhadas

Para usar assinaturas partilhadas com servidores de mensagens, pode especificar o tópico de assinatura partilhada na forma de $shared/<GROUP_NAME>/<TOPIC_FILTER>.

Experiência em operações

Em Detalhes da fonte do fluxo de dados da experiência de operações, selecione Agente de mensagens e use o campo Tópico para especificar o grupo e o tópico de assinatura compartilhada.

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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

Se a contagem de instâncias no perfil de fluxo de dados for maior que uma, a assinatura compartilhada será habilitada automaticamente para todos os fluxos de dados que usam uma fonte de agente de mensagens. Nesse caso, o prefixo $shared é adicionado e o nome do grupo de assinatura compartilhada gerado automaticamente. Por exemplo, se você tiver um perfil de fluxo de dados com uma contagem de instâncias de 3 e seu fluxo de dados usar um ponto de extremidade do agente de mensagens como fonte configurado com tópicos topic1 e topic2, eles serão automaticamente convertidos em assinaturas compartilhadas como $shared/<GENERATED_GROUP_NAME>/topic1 e $shared/<GENERATED_GROUP_NAME>/topic2.

Você pode criar explicitamente um tópico nomeado $shared/mygroup/topic em sua configuração. No entanto, adicionar o $shared tópico explicitamente não é recomendado, pois o prefixo $shared é adicionado automaticamente quando necessário. Os fluxos de dados podem fazer otimizações com o nome do grupo se ele não estiver definido. Por exemplo, $share não está definido e os fluxos de dados só precisam operar sobre o nome do tópico.

Importante

Os fluxos de dados que exigem assinatura compartilhada quando a contagem de instâncias é maior que uma são importantes ao usar o agente MQTT da Grade de Eventos como origem , pois ele não oferece suporte a assinaturas compartilhadas. Para evitar mensagens ausentes, defina a contagem de instâncias do perfil de fluxo de dados como uma ao usar o agente MQTT da Grade de Eventos como origem. É quando o fluxo de dados atua como assinante e está a receber mensagens da nuvem.

Tópicos de Kafka

Quando a origem for um ponto de extremidade Kafka (Hubs de Eventos incluídos), especifique os tópicos Kafka individuais a que subscrever para mensagens recebidas. Não há suporte para curingas, portanto, você deve especificar cada tópico estaticamente.

Observação

Ao usar os Hubs de Eventos através do ponto final Kafka, cada hub de evento individual dentro do espaço de nomes é o tópico Kafka. Por exemplo, se você tiver um namespace Hubs de Eventos com dois hubs thermostats de eventos e humidifiers, poderá especificar cada hub de eventos como um tópico Kafka.

Para configurar os tópicos de Kafka:

Experiência em operações

No fluxo de dados da experiência de operações, nos Detalhes da fonte, selecione Agente de mensagens e utilize o campo Tópico para especificar o filtro de tópicos Kafka a que pretende subscrever para mensagens de entrada.

Observação

Apenas um filtro de tópico pode ser especificado na experiência de operações. Para usar vários filtros de tópico, use Bicep ou Kubernetes.

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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

Especificar esquema de origem

Ao usar MQTT ou Kafka como origem, você pode especificar um esquema para exibir a lista de pontos de dados na interface do usuário da Web da experiência de operações. Atualmente, o uso de um esquema para desserializar e validar mensagens de entrada não é atualmente suportado.

Se a origem for um ativo, o esquema será automaticamente inferido a partir da definição de ativo.

Sugestão

Para gerar o esquema a partir de um arquivo de dados de exemplo, use o Schema Gen Helper.

Para configurar o esquema usado para desserializar as mensagens de entrada de uma fonte:

Experiência em operações

Em Detalhes da fonte do fluxo de dados da experiência de operações, 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, consulte Compreender esquemas de mensagens.

CLI do Azure

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

Bíceps

Depois de usar o registro de esquema para armazenar o esquema, você pode fazer referência a ele na configuração de fluxo de dados.

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

Kubernetes (visualização)

Depois de usar o registro de esquema para armazenar o esquema, você pode fazer referência a ele na configuração de fluxo de dados.

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

Para saber mais, consulte Compreender esquemas de mensagens.

Solicitar persistência de disco

A persistência do disco de solicitação permite que os fluxos de dados mantenham o estado durante as reinicializações. Quando você habilita esse recurso, o gráfico recupera o estado de processamento se o broker conectado for reiniciado. Esse recurso é útil para cenários de processamento com monitoração de estado em que a perda de dados intermediários é um problema. Quando você habilita a persistência de disco de solicitação, o broker persiste os dados MQTT, como mensagens na fila de assinantes, no disco. Essa abordagem garante que a fonte de dados do seu fluxo de dados não perca dados durante quedas de energia ou reinicializações do broker. O broker mantém o desempenho ideal porque a persistência é configurada por fluxo de dados, portanto, apenas os fluxos de dados que precisam de persistência usam esse recurso.

O gráfico de fluxo de dados solicita essa persistência durante a assinatura usando uma propriedade de usuário MQTTv5. Esta funcionalidade funciona apenas quando:

• O fluxo de dados usa o broker ou ativo MQTT como fonte
• O broker MQTT tem persistência habilitada com o modo de persistência dinâmica definido para Enabled o tipo de dados, como filas de assinantes

Essa configuração permite que clientes MQTT, como fluxos de dados, solicitem persistência de disco para suas assinaturas usando as propriedades do usuário MQTTv5. Para obter detalhes sobre a configuração de persistência do MQTT broker, consulte Configure MQTT broker persistence.

A configuração aceita Enabled ou Disabled. Disabled é o padrão.

Experiência em operações

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

CLI do Azure

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

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

Bíceps

Adicione a requestDiskPersistence propriedade ao seu 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
    ]
  }
}

Kubernetes (visualização)

Adicione a requestDiskPersistence propriedade à sua 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

Transformação

A operação de transformação é onde você pode transformar os dados da origem antes de enviá-los para o destino. As transformações são opcionais. Se você não precisar fazer alterações nos dados, não inclua a operação de transformação na configuração do fluxo de dados. Várias transformações são encadeadas em estágios, independentemente da ordem em que são especificadas na configuração. A ordem das etapas é sempre:

1. Enriquecimento: adicione dados adicionais aos dados de origem com um conjunto de dados e uma condição correspondentes.
2. Filtrar: filtre os dados com base em uma condição.
3. Mapear, Calcular, Renomear ou adicionar uma propriedade Novo: mova dados de um campo para outro com uma conversão opcional.

Esta seção é uma introdução às transformações de fluxo de dados. Para obter informações mais detalhadas, consulte Mapear dados usando fluxos de dados, Converter dados usando conversões de fluxo de dados e Enriquecer dados usando fluxos de dados.

Experiência em operações

Na experiência de operações, selecione Fluxo de> dadosAdicionar transformação (opcional).

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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

Enriquecimento: Adicionar dados de referência

Para enriquecer os dados, primeiro adicione o conjunto de dados de referência no repositório de estado do Azure IoT Operations. O conjunto de dados é usado para adicionar dados extras aos dados de origem com base em uma condição. A condição é especificada como um campo nos dados de origem que corresponde a um campo no conjunto de dados.

Você pode carregar dados de exemplo no repositório de estado usando a CLI do repositório de estado. Os nomes das chaves no armazenamento de estado correspondem a um conjunto de dados na configuração do fluxo de dados.

Experiência em operações

Atualmente, o estágio Enrich não é suportado na experiência de operações.

CLI do Azure

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

Bíceps

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

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

Kubernetes (visualização)

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

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

Se o conjunto de dados tiver um registro com o asset campo, semelhante a:

{
  "asset": "thermostat1",
  "location": "room1",
  "manufacturer": "Contoso"
}

Os dados da fonte com o campo deviceId correspondente a thermostat1 têm os campos location e manufacturer disponíveis nas fases de filtro e mapeamento.

Para obter mais informações sobre sintaxe de condição, consulte Enriquecer dados usando fluxos de dados e Converter dados usando fluxos de dados.

Filtrar: filtrar dados com base em uma condição

Para filtrar os dados mediante uma condição, pode usar a etapa filter. A condição é especificada como um campo nos dados de origem que corresponde a um valor.

Experiência em operações

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

   

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

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

   No campo de condição do 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 $metadata sintaxe só é necessária para propriedades MQTT que fazem parte do cabeçalho da mensagem. Para obter mais informações, consulte referências de campo.

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

3. Selecione Aplicar.

CLI do Azure

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

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

Bíceps

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

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

Se o temperature campo for maior que 20, os dados são passados para a próxima etapa. Se o temperature campo for menor ou igual a 20, os dados serão filtrados.

Kubernetes (visualização)

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

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

Se o temperature campo for maior que 20, os dados são passados para a próxima etapa. Se o temperature campo for menor ou igual a 20, os dados serão filtrados.

Mapa: Mover dados de um campo para outro

Para mapear os dados para outro campo com conversão opcional, você pode usar a map operação. A conversão é especificada como uma fórmula que usa os campos nos dados de origem.

Na experiência de operações, o mapeamento é atualmente suportado usando as transformações Compute, Rename e Propriedade nova.

Computação

Você pode usar a transformação Compute para aplicar uma fórmula aos dados de origem. Esta operação é usada para aplicar uma fórmula aos dados de origem e armazenar o campo de resultado.

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

   

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

   Configurações	Descrição
   Selecionar fórmula	Escolha uma fórmula existente na lista pendente ou selecione Personalizar para introduzir uma fórmula manualmente.
   Resultado	Especifique o nome de exibição da saída para o resultado.
   Fórmula	Insira a fórmula a ser aplicada aos dados de origem.
   Descrição	Forneça uma descrição para a transformação.
   Último valor conhecido	Opcionalmente, use o último valor conhecido se o valor atual não estiver disponível.

   Pode introduzir ou editar uma fórmula no campo Fórmula . A fórmula pode usar os campos nos dados de origem. Insira @ ou selecione Ctrl + Espaço para escolher pontos de dados em uma lista suspensa. Para fórmulas incorporadas, selecione o espaço reservado <dataflow> para ver a lista de dados disponíveis.

   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 $metadata sintaxe só é necessária para propriedades MQTT que fazem parte do cabeçalho da mensagem. Para obter mais informações, consulte referências de campo.

   A fórmula pode usar os campos nos dados de origem. Por exemplo, você pode usar o temperature campo nos dados de origem para converter a temperatura em Celsius e armazená-la no temperatureCelsius campo de saída.

3. Selecione Aplicar.

Mudar o nome

Você pode renomear um ponto de dados usando a transformação Renomear . Esta operação é usada para renomear um ponto de dados nos dados de origem para um novo nome. O novo nome pode ser usado nos estágios subsequentes do fluxo de dados.

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

   

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

   Configurações	Descrição
   Ponto de dados	Selecione um ponto de dados na lista de opções ou insira um cabeçalho $metadata.
   Novo nome de ponto de dados	Insira o novo nome para o ponto de dados.
   Descrição	Forneça uma descrição para a transformação.

   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 $metadata sintaxe só é necessária para propriedades MQTT que fazem parte do cabeçalho da mensagem. Para obter mais informações, consulte referências de campo.

3. Selecione Aplicar.

Nova propriedade

Você pode adicionar uma nova propriedade aos dados de origem usando a transformação New propriedade. Esta operação é usada para adicionar uma nova propriedade aos dados de origem. A nova propriedade pode ser usada nos estágios subsequentes do fluxo de dados.

Experiência em operações

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

   

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

   Configurações	Descrição
   Chave da propriedade	Insira a chave da nova propriedade.
   Valor do imóvel	Insira o valor da nova propriedade.
   Descrição	Forneça uma descrição para a nova propriedade.

3. Selecione Aplicar.

CLI do Azure

Por exemplo, você pode usar o temperature campo nos dados de origem para converter a temperatura em Celsius e armazená-la no temperatureCelsius campo. Você também pode enriquecer os dados de origem com o location campo 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"
      }
    ]
  }
}

Bíceps

Você pode acessar as 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>. Para obter mais informações, consulte referências de campo.

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

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

Kubernetes (visualização)

Você pode acessar as 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>. Para obter mais informações, consulte referências de campo.

Por exemplo, você pode usar o temperature campo nos dados de origem para converter a temperatura em Celsius e armazená-la no temperatureCelsius campo. Você também pode enriquecer os dados de origem com o location campo 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

Para saber mais, consulte Mapear dados usando fluxos de dados e Converter dados usando fluxos de dados.

Eliminar

Por padrão, todos os pontos de dados são incluídos no esquema de saída. Você pode remover qualquer ponto de dados do destino usando a transformação Remover .

Experiência em operações

1. Em Transformar (opcional), selecione Remover.

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

   

3. Selecione Aplicar.

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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 saber mais, consulte Mapear dados usando fluxos de dados e Converter dados usando fluxos de dados.

Serializar dados de acordo com um esquema

Se desejar serializar os dados antes de enviá-los para o destino, será necessário especificar um esquema e um formato de serialização. Caso contrário, os dados são serializados em JSON com os tipos inferidos. Pontos de extremidade de armazenamento como o Microsoft Fabric ou o Azure Data Lake exigem um esquema para garantir a consistência dos dados. Os formatos de serialização suportados são Parquet e Delta.

Sugestão

Para gerar o esquema a partir de um arquivo de dados de exemplo, use o Schema Gen Helper.

Experiência em operações

Na experiência em 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 oferecem suporte a formatos de serialização são Microsoft Fabric OneLake, Azure Data Lake Storage Gen 2, Azure Data Explorer e armazenamento local. Por exemplo, para serializar os dados no formato Delta, você precisa carregar um esquema no registro do esquema e fazer referência a ele na configuração do ponto de extremidade de destino do fluxo de dados.

CLI do Azure

Depois de carregar um esquema no registro do esquema, você pode fazer referência a ele na configuração do fluxo de dados.

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

Bíceps

Depois de carregar um esquema no registro do esquema, você pode fazer referência a ele na configuração do fluxo de dados.

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

Kubernetes (visualização)

Depois de carregar um esquema no registro do esquema, você pode fazer referência a ele na configuração do fluxo de dados.

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

Para obter mais informações sobre o registro de esquema, consulte Compreender esquemas de mensagem.

Destino

Para configurar um destino para o fluxo de dados, especifique a referência do ponto de extremidade e o destino dos dados. Você pode especificar uma lista de destinos de dados para o endereço de destino.

Para enviar dados para um destino diferente do broker MQTT local, crie um ponto de extremidade de fluxo de dados. Para saber como, consulte Configurar pontos de extremidade de fluxo de dados. Se o destino não for o broker MQTT local, ele deverá ser usado como origem. Para saber mais sobre como usar o ponto de extremidade do corretor MQTT local, consulte Os fluxos de dados devem usar o ponto de extremidade do corretor MQTT local.

Importante

Os pontos de extremidade de armazenamento exigem um esquema para serialização. Para usar o fluxo de dados com o Microsoft Fabric OneLake, o Azure Data Lake Storage, o Azure Data Explorer ou o Armazenamento Local, você deve especificar uma referência de esquema.

Experiência em operações

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

   

   Os pontos de extremidade de armazenamento exigem um esquema para serialização. Se você escolher um Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer ou ponto de extremidade de destino do Armazenamento Local, deverá especificar uma referência de esquema. Por exemplo, para serializar os dados para um ponto de extremidade do Microsoft Fabric no formato Delta, você precisa carregar um esquema no registro do esquema e fazer referência a ele 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 a qual enviar os dados. Consulte Configurar destino de dados (tópico, contêiner ou tabela) para obter mais informações.

CLI do Azure

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

Bíceps

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

Kubernetes (visualização)

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

Configurar destino de dados (tópico, contêiner ou tabela)

Semelhante às fontes de dados, o destino de dados é um conceito usado para manter os pontos de extremidade de fluxo de dados reutilizáveis em vários fluxos de dados. Essencialmente, ele representa o subdiretório na configuração do ponto de extremidade do fluxo de dados. Por exemplo, se o ponto de extremidade do fluxo de dados for um ponto de extremidade de armazenamento, o destino dos dados será a tabela na conta de armazenamento. Se o ponto de extremidade do fluxo de dados for um ponto de extremidade Kafka, o destino dos dados será o tópico Kafka.

Tipo de ponto final	Significado do destino dos dados	Descrição
MQTT (ou Event Grid)	Tópico	O tópico MQTT para onde os dados são enviados. Suporta tópicos estáticos e tradução dinâmica de tópicos usando variáveis como ${inputTopic} e ${inputTopic.index}. Para obter mais informações, consulte Tópicos de destino dinâmico.
Kafka (ou Hubs de Eventos)	Tópico	O tópico Kafka para onde os dados são enviados. Apenas tópicos estáticos são suportados, sem curingas. Se o ponto de extremidade for um namespace de Hubs de Eventos, o destino dos dados será o hub de eventos individual contido no namespace.
Azure Data Lake Storage	Contentor	O contêiner na conta de armazenamento. Não a tabela.
Microsoft Fabric OneLake	Tabela ou pasta	Corresponde ao tipo de caminho configurado para o ponto de extremidade.
Azure Data Explorer	Tabela	A tabela no banco de dados do Azure Data Explorer.
Armazenamento Local	Pasta de Arquivos	O nome da pasta ou diretório na montagem de volume persistente de armazenamento local. Ao usar o Armazenamento de Contêiner do Azure habilitado pelo Azure Arc Cloud Ingest Edge Volumes, isso deve corresponder ao spec.path parâmetro para o subvolume que você criou.
OpenTelemetry	Tópico	O tópico OpenTelemetry para o qual os dados são enviados. Apenas tópicos estáticos são suportados.

Para configurar o destino dos dados:

Experiência em operações

Ao usar a experiência de operações, o campo de destino de dados é interpretado automaticamente com base no tipo de ponto final. Por exemplo, se o ponto de extremidade do 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 do 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.

CLI do Azure

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

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

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

Ou, se tiver um ponto de hub de eventos personalizado, a configuração terá a seguinte aparência:

{
  "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 encaminhar mensagens de factory/1/data para processed/factory/1:

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

Bíceps

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 para o broker MQTT local um tópico MQTT estático, use a seguinte configuração:

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

Ou, se tiver um ponto de hub de eventos personalizado, a configuração terá a seguinte aparência:

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

Kubernetes (visualização)

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 para o broker MQTT local um tópico MQTT estático, use a seguinte configuração:

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Ou, se tiver um ponto de hub de eventos personalizado, a configuração terá a seguinte aparência:

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}

Tópicos de destino dinâmico

Para pontos de extremidade MQTT, você pode usar variáveis de tópico dinâmico no dataDestination campo para rotear mensagens com base na estrutura do tópico de origem. Estão disponíveis as seguintes variáveis:

• ${inputTopic} - O tópico de entrada original completo
• ${inputTopic.index} - Um segmento do tópico de entrada (índice começa em 1)

Por exemplo, processed/factory/${inputTopic.2} roteia mensagens de factory/1/data para processed/factory/1. Os segmentos de tópico são indexados por 1 e as barras à esquerda/à direita são ignoradas.

Se uma variável de tópico não puder ser resolvida (por exemplo, ${inputTopic.5} quando o tópico de entrada tiver apenas três segmentos), a mensagem será descartada e um aviso será registrado. Caracteres curinga (# e +) não são permitidos nos tópicos de destino.

Observação

Os caracteres $, {e } são válidos em nomes de tópicos MQTT, portanto, um tópico como factory/$inputTopic.2 é aceitável, mas incorreto se você pretendia usar a variável de tópico dinâmico.

Exemplo

O exemplo a seguir é uma configuração de fluxo de dados que usa o ponto de extremidade MQTT para a origem e o destino. A fonte filtra os dados do tópico azure-iot-operations/data/thermostatMQTT. A transformação converte a temperatura em Fahrenheit e filtra os dados onde a temperatura multiplicada pela humidade é inferior a 100000. O destino envia os dados para o tópico MQTT factory.

Experiência em operações

CLI do Azure

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 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 inicial 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 exemplo de comando 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 dinâmica de tópicos para rotear mensagens de diferentes termostatos 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.

Bíceps

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

Kubernetes (visualização)

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

Para ver mais exemplos de configurações de fluxo de dados, consulte API REST do Azure - Fluxo de dados e o Bicep de início rápido.

Verificar se um fluxo de dados está funcionando

Siga o tutorial: Ponte MQTT bidirecional para a Grelha de Eventos do Azure para verificar se o fluxo de dados está a funcionar.

Exportar configuração de fluxo de dados

Para exportar a configuração de fluxo de dados, pode utilizar a experiência operacional ou exportar o recurso personalizado do fluxo de dados.

Experiência em operações

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

CLI do Azure

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 exemplo de comando 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

Bíceps

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

Kubernetes (visualização)

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

Configuração adequada do fluxo de dados

Para garantir que o fluxo de dados esteja funcionando conforme o esperado, verifique o seguinte:

• O ponto de extremidade de fluxo de dados MQTT padrão deve ser usado como origem ou destino.
• O perfil de fluxo de dados existe e é referenciado na configuração de fluxo de dados.
• Source é um endpoint MQTT, um endpoint Kafka ou um ativo. Os pontos finais de armazenamento não podem ser usados como origem.
• Ao usar a Grade de Eventos como origem, a contagem de instâncias do perfil de fluxo de dados é definida como 1 porque o agente MQTT da Grade de Eventos não oferece suporte a assinaturas compartilhadas.
• Ao usar Hubs de Eventos como origem, cada hub de eventos no namespace é um tópico Kafka separado e deve ser especificado como a fonte de dados.
• A transformação, se utilizada, é configurada com a sintaxe adequada, incluindo o escape adequado de caracteres especiais.
• Ao usar pontos de extremidade de tipo de armazenamento como destino, um esquema é especificado.
• Ao usar tópicos de destino dinâmico para pontos de extremidade MQTT, certifique-se de que as variáveis de tópico façam referência a segmentos válidos.

Próximos passos

• Mapeie dados usando fluxos de dados
• Converter dados usando fluxos de dados
• Enriqueça dados usando fluxos de dados
• Compreender esquemas de mensagens
• Gerenciar perfis de fluxo de dados