Configurar fluxos de dados no Azure IoT Operations

Para criar um fluxo de dados na experiência de operações, selecione Fluxo de dados>Criar fluxo de dados. Em seguida, você verá a página em que poderá configurar a fonte, a transformação e o destino do fluxo de dados.

Crie um arquivo .bicep do Bicep para começar a criar um fluxo de dados. Esse exemplo mostra a estrutura do fluxo de dados que contém as configurações de fonte, 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-08-15-preview' existing = {
  name: aioInstanceName
}

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

resource defaultDataflowEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-08-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

// Pointer to the default dataflow profile
resource defaultDataflowProfile 'Microsoft.IoTOperations/instances/dataflowProfiles@2024-08-15-preview' existing = {
  parent: aioInstance
  name: 'default'
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2024-08-15-preview' = {
  // Reference to the parent dataflow profile, the default profile in this case
  // Same usage as profileRef in Kubernetes YAML
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    operations: [
      {
        operationType: 'Source'
        sourceSettings: {
          // See source configuration section
        }
      }
      {
        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. Esse exemplo mostra a estrutura do fluxo de dados que contém as configurações de fonte, transformação e destino.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  # Reference to the default dataflow profile
  # This field is required when configuring via Kubernetes YAML
  # The syntax is different when using Bicep
  profileRef: default 
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        # See source configuration section
    - operationType: BuiltInTransformation
      builtInTransformationSettings:
        # See transformation configuration section
    - operationType: Destination
      destinationSettings:
        # See destination configuration section

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

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

   

2. Insira as seguintes configurações para a fonte MQTT:

   Configuração	Descrição
   Tópico do MQTT	O filtro de tópico MQTT a ser assinado para as mensagens de entrada. Confira Configurar tópicos do MQTT ou do 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.

O ponto de extremidade MQTT está configurado no arquivo de modelo do Bicep. Por exemplo, o seguinte ponto de extremidade é uma fonte para o fluxo de dados.

sourceSettings: {
  endpointRef: defaultDataflowEndpoint
  dataSources: [
    'thermostats/+/telemetry/temperature/#',
    'humidifiers/+/telemetry/humidity/#'
  ]
}

Aqui, dataSources permite que você especifique vários tópicos MQTT ou Kafka sem precisar modificar a configuração do ponto de extremidade. Isso significa que o mesmo ponto de extremidade pode ser reutilizado em vários fluxos de dados, mesmo que os tópicos variem. Para saber mais, confira Configurar fontes de dados.

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

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

Como dataSources permite que você especifique 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, confira Configurar fontes de dados.

No momento, não há suporte para o uso de um ponto de extremidade MQTT ou Kafka personalizado como fonte na experiência de operações.

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

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

Nos Detalhes da fonte de fluxo de dados da experiência de operações, selecione MQTT e, em seguida, use o campo Tópico MQTT para especificar o filtro de tópico MQTT a ser assinado para as mensagens de entrada.

sourceSettings: {
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources: [
    '<MQTT_TOPIC_FILTER_1>',
    '<MQTT_TOPIC_FILTER_2>'
    // Add more MQTT topic filters as needed
  ]
}

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

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

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

sourceSettings:
  endpointRef: <MQTT_ENDPOINT_NAME>
  dataSources:
    - <MQTT_TOPIC_FILTER_1>
    - <MQTT_TOPIC_FILTER_2>
    # Add more MQTT topic filters as needed

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

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

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

No momento, não há suporte para o uso de um ponto de extremidade do Kafka como uma fonte na experiência de operações.

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

Nos Detalhes da fonte de fluxo de dados da experiência de operações, selecione MQTT 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 Entenda os esquemas de mensagens.

Depois de usar o registro de esquema para armazenar o esquema, é possível 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, é possível referenciá-lo na configuração do fluxo de dados.

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

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

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

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

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

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, a operação de enriquecimento não está disponível na experiência de operações.

Esse exemplo mostra como você pode usar o campo deviceId nos dados de origem para corresponder ao campo asset 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 campo asset, 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 nos estágios de filtro e mapa.

Por exemplo, você pode usar o campo deviceId nos dados de origem para corresponder ao campo asset 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 campo asset, 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 nos estágios de filtro e mapa.

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

2. Escolha os pontos de dados a serem incluídos no conjunto de dados.

3. Adicione uma condição de filtro e uma descrição.

   

4. Escolha Aplicar.

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.

Por exemplo, você pode usar o campo temperature nos dados de origem 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 de origem 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.

Na experiência de operações, atualmente há suporte para mapeamento usando transformações de computação.

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

2. Insira os campos e expressões obrigatórios.

   

3. Escolha Aplicar.

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

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

Atualmente, não há suporte para a especificação do esquema de saída e da serialização na experiência de operações.

Depois de carregar um esquema no registro de esquemas, será possível 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 esquemas, será possível 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 do fluxo de dados a ser usado como destino.

   

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

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

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
}

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

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

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

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