Создание потоков данных в операциях Интернета вещей Azure

1. Чтобы создать поток данных в рабочей среде, выберите Поток данных>.

2. Выберите имя заполнителя new-data-flow , чтобы задать свойства потока данных. Введите имя потока данных и выберите используемый профиль потока данных. Профиль потока данных по умолчанию выбирается по умолчанию. Дополнительные сведения о профилях потока данных см. в разделе "Настройка профиля потока данных".

   

   Внимание

   При создании потока данных можно выбрать только профиль потока данных. После создания потока данных невозможно изменить профиль потока данных. Если вы хотите изменить профиль потока данных существующего потока данных, удалите исходный поток данных и создайте новый с новым профилем потока данных.

3. Настройте исходную, преобразование и целевую конечную точку для потока данных, выбрав элементы на схеме потока данных.

   

Создайте файл Bicep .bicep , чтобы начать создание потока данных. В этом примере показана структура потока данных, содержащего конфигурации источника, преобразования и назначения.

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

Создайте файл манифеста .yaml Kubernetes, чтобы начать создание потока данных. В этом примере показана структура потока данных, содержащего конфигурации источника, преобразования и назначения.

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. В разделе "Исходные сведения" выберите брокер сообщений.

   

2. Введите следующие параметры для источника брокера сообщений:

   Настройка	Описание
   Конечная точка потока данных	Выберите по умолчанию, чтобы использовать стандартную конечную точку брокера сообщений MQTT.
   Тема	Фильтр раздела для подписки на входящие сообщения. Используйте Тему(ы)>Добавить строку, чтобы добавить несколько тем. Дополнительные сведения о разделах см. в разделе "Настройка MQTT" или "Kafka".
   Схема сообщений	Схема, используемая для десериализации входящих сообщений. См. раздел "Указание схемы для десериализации данных".

3. Выберите Применить.

Конечная точка брокера сообщений настраивается в файле Bicep. Например, следующая конечная точка является источником потока данных.

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

dataSources Здесь можно указать несколько разделов MQTT или Kafka, не изменяя конфигурацию конечной точки. Эта гибкость означает, что одна конечная точка может использоваться повторно в нескольких потоках данных, даже если разделы различаются. Дополнительные сведения см. в разделе "Настройка источников данных".

Например, чтобы настроить источник с помощью конечной точки брокера сообщений и двух фильтров разделов, используйте следующую конфигурацию:

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

Так как dataSources можно указать разделы MQTT или Kafka без изменения конфигурации конечной точки, можно повторно использовать конечную точку для нескольких потоков данных, даже если разделы отличаются. Дополнительные сведения см. в разделе "Настройка источников данных".

Вы можете использовать ресурс в качестве источника для потока данных. Использование актива в качестве источника доступно только в интерфейсе операций.

1. В разделе "Исходные сведения" выберите "Ресурс".

2. Выберите ресурс, который вы хотите использовать в качестве исходной конечной точки.

3. Выберите Продолжить.

   Отображается список точек данных для выбранного ресурса.

   

4. Выберите "Применить" , чтобы использовать ресурс в качестве исходной конечной точки.

Настройка ресурса в качестве источника доступна только в интерфейсе операций.

Настройка ресурса в качестве источника доступна только в интерфейсе операций.

1. В разделе "Исходные сведения" выберите брокер сообщений.

   

2. Введите следующие параметры для источника брокера сообщений:

   Настройка	Описание
   Конечная точка потока данных	Нажмите кнопку повторного выбора , чтобы выбрать пользовательскую конечную точку потока данных MQTT или Kafka. Дополнительные сведения см. в разделе Настройка конечных точек потока данных MQTT или Настройка конечных точек потока данных Azure Event Hubs и Kafka.
   Тема	Фильтр раздела для подписки на входящие сообщения. Используйте Тему(ы)>Добавить строку, чтобы добавить несколько тем. Дополнительные сведения о разделах см. в разделе "Настройка MQTT" или "Kafka".
   Схема сообщений	Схема, используемая для десериализации входящих сообщений. См. раздел "Указание схемы для десериализации данных".

3. Выберите Применить.

Замените значения заполнителей именем и темами настраиваемой конечной точки.

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

В сведениях о источнике потока данных для операций выберите брокер сообщений, а затем используйте поле "Темы", чтобы указать фильтры разделов MQTT для подписки на входящие сообщения. Вы можете добавить несколько разделов MQTT, выбрав "Добавить строку " и введя новый раздел.

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

Пример с несколькими фильтрами разделов MQTT с подстановочными знаками:

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

Здесь подстановочный знак + используется для выбора всех устройств в темах thermostats и humidifiers. Глобальный символ # используется для выбора всех сообщений телеметрии во всех подтопиках тем temperature и humidity.

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

Пример с несколькими фильтрами разделов с подстановочными знаками:

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

Здесь подстановочный знак + используется для выбора всех устройств в темах thermostats и humidifiers. Глобальный символ # используется для выбора всех сообщений телеметрии во всех подтопиках тем temperature и humidity.

В данных о потоке операций о сведениях источника выберите Брокер сообщений и используйте поле "Тема", чтобы указать группу совместной подписки и тему.

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

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

В данных об опыте операций Source details выберите Message broker, а затем используйте поле Топик, чтобы указать фильтр топика Kafka для подписки на входящие сообщения.

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: {
  serializationFormat: 'Json'
  schemaRef: 'aio-sr://<SCHEMA_NAMESPACE>/<SCHEMA_NAME>:<VERSION>'
}

После использования реестра схем для хранения схемы вы можете ссылаться на нее в конфигурации потока данных.

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

В интерфейсе операций выберите "Добавить преобразование потока>данных" (необязательно).

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

В настоящее время этап обогащения не поддерживается в интерфейсе операций.

В этом примере показано, как можно использовать deviceId поле в исходных данных для сопоставления asset поля в наборе данных:

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

Например, можно использовать deviceId поле в исходных данных для сопоставления asset поля в наборе данных:

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

1. В разделе «Преобразование (необязательно)» выберите «Фильтр»>«Добавить».

   

2. Введите необходимые параметры.

   Настройка	Описание
   Условие фильтра	Условие фильтрации данных на основе поля в исходных данных.
   Описание	Укажите описание условия фильтра.

   В поле условия фильтра введите значение @ или нажмите CTRL+ПРОБЕЛ, чтобы выбрать точки данных из выпадающего списка.

   Вы можете ввести свойства метаданных MQTT с помощью формата @$metadata.user_properties.<property> или @$metadata.topic. Вы также можете ввести заголовки $metadata с помощью формата @$metadata.<header>. Синтаксис $metadata необходим только для свойств MQTT, входящих в заголовок сообщения. Для получения дополнительной информации см. справочную информацию о полях.

   Условие может использовать поля в исходных данных. Например, можно использовать условие фильтрации, такое как @temperature > 20, чтобы отфильтровать данные, равные или меньшие 20, на основе поля температуры.

3. Выберите Применить.

Например, можно использовать temperature поле в исходных данных для фильтрации данных:

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

temperature Если поле больше 20, данные передаются на следующий этап. temperature Если поле меньше или равно 20, данные фильтруются.

Например, можно использовать temperature поле в исходных данных для фильтрации данных:

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

temperature Если поле больше 20, данные передаются на следующий этап. temperature Если поле меньше или равно 20, данные фильтруются.

1. В разделе "Преобразование" (необязательно) выберите "Добавить новое свойство>".

   

2. Введите необходимые параметры.

   Настройка	Описание
   Ключ свойства	Введите ключ для нового свойства.
   Значение свойства	Введите значение нового свойства.
   Описание	Укажите описание нового свойства.

3. Выберите Применить.

Вы можете получить доступ к свойствам метаданных MQTT с помощью формата $metadata.user_properties.<property> или $metadata.topic. Вы также можете ввести заголовки $metadata с помощью формата $metadata.<header>. Для получения дополнительной информации см. справочную информацию о полях.

Например, можно использовать temperature поле в исходных данных для преобразования температуры в Цельсию и хранения его в temperatureCelsius поле. Вы также можете дополнить исходные данные location полем из набора данных контекстуализации:

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

Вы можете получить доступ к свойствам метаданных MQTT с помощью формата $metadata.user_properties.<property> или $metadata.topic. Вы также можете ввести заголовки $metadata с помощью формата $metadata.<header>. Для получения дополнительной информации см. справочную информацию о полях.

Например, можно использовать temperature поле в исходных данных для преобразования температуры в Цельсию и хранения его в temperatureCelsius поле. Вы также можете дополнить исходные данные location полем из набора данных контекстуализации:

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

1. В разделе "Преобразование( необязательно)" нажмите кнопку "Удалить".

2. Выберите точку данных, чтобы удалить из выходной схемы.

   

3. Выберите Применить.

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

Для опыта работы с операциями вы указываете формат схемы и сериализации в деталях конечной точки потока данных. Конечные точки, поддерживающие форматы сериализации, — Microsoft Fabric OneLake, Azure Data Lake Storage 2-го поколения, Azure Data Explorer и локальное хранилище. Например, чтобы сериализовать данные в разностном формате, необходимо отправить схему в реестр схем и ссылаться на нее в конфигурации конечной точки назначения потока данных.

После отправки схемы в реестр схем можно ссылаться на нее в конфигурации потока данных.

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

После отправки схемы в реестр схем можно ссылаться на нее в конфигурации потока данных.

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

1. Выберите конечную точку потока данных, используемую в качестве назначения.

   

   Для конечных точек хранилища требуется схема сериализации. Если выбрать конечную точку назначения Microsoft Fabric OneLake, Azure Data Lake Storage, Azure Data Explorer или конечную точку назначения локального хранилища, необходимо указать ссылку на схему. Например, чтобы сериализовать данные в конечную точку Microsoft Fabric в формате Delta, необходимо добавить схему в реестр схем и ссылаться на нее в конфигурации конечной точки потока данных.

   

2. Нажмите кнопку "Продолжить", чтобы настроить назначение.

3. Введите необходимые параметры для назначения, включая раздел или таблицу для отправки данных. Дополнительные сведения см. в разделе "Настройка назначения данных" (раздел, контейнер или таблица).

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

При использовании интерфейса операций поле назначения данных автоматически интерпретируется на основе типа конечной точки. Например, если конечная точка потока данных является конечной точкой хранения, страница сведений о назначении предложит ввести имя контейнера. Если конечная точка потока данных является конечной точкой MQTT, страница сведений о назначении предложит ввести тему.

Синтаксис одинаков для всех конечных точек потока данных:

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

Например, чтобы отправить данные обратно в локальный брокер MQTT по статической теме MQTT, используйте следующую конфигурацию.

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

Или, если у вас есть пользовательская конечная точка концентратора событий, конфигурация будет выглядеть следующим образом:

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

Другой пример использования конечной точки хранилища в качестве назначения:

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

Синтаксис одинаков для всех конечных точек потока данных:

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

Например, чтобы отправить данные обратно в локальный брокер MQTT по статической теме MQTT, используйте следующую конфигурацию.

destinationSettings:
  endpointRef: default
  dataDestination: example-topic

Или, если у вас есть пользовательская конечная точка концентратора событий, конфигурация будет выглядеть следующим образом:

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

Другой пример использования конечной точки хранилища в качестве назначения:

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

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

Выберите поток данных, который вы хотите экспортировать, и выберите " Экспорт " на панели инструментов.

Bicep — это инфраструктура в виде кода, и экспорт не требуется. Используйте файл Bicep для создания потока данных для быстрого настройки и настройки потоков данных.

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