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

Внимание

Предварительная версия операций Интернета вещей Azure, включенная Azure Arc в настоящее время в предварительной версии. Не следует использовать это программное обеспечение предварительной версии в рабочих средах.

Вам потребуется развернуть новую установку Операций Интернета вещей Azure, когда общедоступная версия станет доступной. Вы не сможете обновить предварительную установку.

Юридические условия, применимые к функциям Azure, которые находятся в состоянии бета-версии, предварительной версии или иным образом еще не выпущены в общедоступной версии, см. на странице Дополнительные условия использования предварительных версий в Microsoft Azure.

Поток данных — это путь, который данные принимают из источника в место назначения с необязательными преобразованиями. Поток данных можно настроить, создав пользовательский ресурс потока данных или используя портал Azure IoT Operations Studio. Поток данных состоит из трех частей: источника, преобразования и назначения.

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

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

Необходимые компоненты

• Экземпляр предварительной версии операций Интернета вещей Azure
• Настроенный профиль потока данных
• Конечные точки потока данных. Например, создайте конечную точку потока данных для локального брокера MQTT. Эту конечную точку можно использовать как для источника, так и для назначения. Кроме того, вы можете попробовать другие конечные точки, такие как Kafka, Центры событий или Azure Data Lake Storage. Сведения о настройке каждого типа конечной точки потока данных см. в разделе "Настройка конечных точек потока данных".

Создание потока данных

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

Портал

Чтобы создать поток данных на портале операций, выберите >"Создать поток данных".

Kubernetes

Общая структура конфигурации потока данных выглядит следующим образом:

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  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

Ознакомьтесь со следующими разделами, чтобы узнать, как настроить типы операций потока данных.

Настройка источника с конечной точкой потока данных для получения данных

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

Портал

Использование ресурса в качестве источника

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

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

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

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

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

   

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

Kubernetes

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

Использование MQTT в качестве источника

Портал

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

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

3. Выберите схему сообщения из раскрывающегося списка или отправьте новую схему. Если исходные данные имеют необязательные поля или поля с различными типами, укажите схему десериализации, чтобы обеспечить согласованность. Например, данные могут содержать поля, которые не присутствуют во всех сообщениях. Без схемы преобразование не может обрабатывать эти поля, так как они будут иметь пустые значения. С помощью схемы можно указать значения по умолчанию или игнорировать поля.

   

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

Kubernetes

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

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

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

Указание схемы для десериализации данных

Если исходные данные имеют необязательные поля или поля с различными типами, укажите схему десериализации, чтобы обеспечить согласованность. Например, данные могут содержать поля, которые не присутствуют во всех сообщениях. Без схемы преобразование не может обрабатывать эти поля, так как они будут иметь пустые значения. С помощью схемы можно указать значения по умолчанию или игнорировать поля.

spec:
  operations:
  - operationType: Source
    sourceSettings:
      serializationFormat: Json
      schemaRef: aio-sr://exampleNamespace/exampleAvroSchema:1.0.0

Чтобы указать схему, создайте файл и сохраните его в реестре схем.

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperature", "type": "float"}
  ]
}

Примечание.

Единственным поддерживаемым форматом сериализации является JSON. Схема является необязательной.

Дополнительные сведения о реестре схем см. в разделе "Общие сведения о схемах сообщений".

Общие подписки

Чтобы использовать общие подписки с источниками MQTT, можно указать раздел общей подписки в виде $shared/<subscription-group>/<topic>.

sourceSettings:
  dataSources:
    - $shared/myGroup/thermostats/+/telemetry/temperature/#

Примечание.

Если число экземпляров в профиле потока данных больше 1, необходимо использовать раздел общей подписки.

Настройка преобразования для обработки данных

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

1. Обогащение. Добавьте дополнительные данные в исходные данные, заданные набором данных и условием для сопоставления.
2. Фильтр. Фильтрация данных на основе условия.
3. Карта. Перемещение данных из одного поля в другое с необязательным преобразованием.

Портал

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

Kubernetes

builtInTransformationSettings:
  datasets:
    # ...
  filter:
    # ...
  map:
    # ...

Обогащение: добавление ссылочных данных

Для обогащения данных можно использовать эталонный набор данных в распределенном хранилище состояний Операций Интернета вещей Azure (DSS). Набор данных используется для добавления дополнительных данных в исходные данные на основе условия. Условие указывается в качестве поля в исходных данных, которые соответствуют полю в наборе данных.

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

Портал

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

Kubernetes

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

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

Если набор данных содержит запись с asset полем, аналогично:

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

Данные из источника с deviceId сопоставлением thermostat1 полей имеют доступные location manufacturer и filter map этапы поля.

Образец данных можно загрузить в DSS с помощью примера средства набора DSS.

Дополнительные сведения о синтаксисе условий см. в разделе "Обогащение данных с помощью потоков данных" и "Преобразование данных с помощью потоков данных".

Фильтр: фильтрация данных на основе условия

Чтобы отфильтровать данные по условию filter , можно использовать этап. Условие указывается в качестве поля в исходных данных, которые соответствуют значению.

Портал

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

2. Выберите точки данных для включения в набор данных.

3. Добавьте условие фильтра и описание.

   

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

Kubernetes

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

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

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

Карта: перемещение данных из одного поля в другое

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

Портал

На портале операций сопоставление в настоящее время поддерживается с помощью преобразований вычислений .

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

2. Введите обязательные поля и выражения.

   

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

Kubernetes

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

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

Дополнительные сведения см. в статье "Сопоставление данных с помощью потоков данных" и "Преобразование данных с помощью потоков данных".

Сериализация данных в соответствии со схемой

Если необходимо сериализовать данные перед отправкой в место назначения, необходимо указать формат схемы и сериализации. В противном случае данные сериализуются в ФОРМАТЕ JSON с выводом типов. Помните, что конечные точки хранения, такие как Microsoft Fabric или Azure Data Lake, требуют схемы для обеспечения согласованности данных.

Портал

Укажите схему вывода при добавлении конечной точки потока данных назначения.

Kubernetes

builtInTransformationSettings:
  serializationFormat: Parquet
  schemaRef: aio-sr://<NAMESPACE>/<SCHEMA>:<VERSION>

Чтобы указать схему, можно создать пользовательский ресурс схемы с определением схемы.

Дополнительные сведения о реестре схем см. в разделе "Общие сведения о схемах сообщений".

{
  "type": "record",
  "name": "Temperature",
  "fields": [
    {"name": "deviceId", "type": "string"},
    {"name": "temperatureCelsius", "type": "float"}
    {"name": "location", "type": "string"}
  ]
}

Поддерживаемые форматы сериализации : JSON, Parquet и Delta.

Настройка назначения с конечной точкой потока данных для отправки данных

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

Портал

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

   

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

3. Добавьте сведения о сопоставлении на основе типа назначения.

Kubernetes

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

destinationSettings:
  endpointRef: mq
  dataDestination: factory

Если вы создали такие конечные точки хранилища, как Microsoft Fabric, используйте поле назначения данных для указания имени таблицы или контейнера:

destinationSettings:
  endpointRef: adls
  dataDestination: telemetryTable

Пример

В следующем примере показана конфигурация потока данных, которая использует конечную точку MQTT для источника и назначения. Источник фильтрует данные из разделов thermostats/+/telemetry/temperature/# MQTT и humidifiers/+/telemetry/humidity/#. Преобразование преобразует температуру в Fahrenheit и фильтрует данные, в которых температура меньше 100000. Назначение отправляет данные в раздел factoryMQTT.

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: my-dataflow
  namespace: azure-iot-operations
spec:
  profileRef: default
  mode: Enabled
  operations:
    - operationType: Source
      sourceSettings:
        endpointRef: mq
        dataSources:
          - thermostats/+/telemetry/temperature/#
          - humidifiers/+/telemetry/humidity/#
    - operationType: builtInTransformation
      builtInTransformationSettings:
        filter:
          - inputs:
              - 'temperature.Value'
              - '"Tag 10".Value'
            expression: "$1*$2<100000"
        map:
          - inputs:
              - '*'
            output: '*'
          - inputs:
              - temperature.Value
            output: TemperatureF
            expression: cToF($1)
          - inputs:
              - '"Tag 10".Value'
            output: 'Tag 10'
    - operationType: Destination
      destinationSettings:
        endpointRef: mq
        dataDestination: factory

Проверка работы потока данных

Следуйте инструкциям из руководства. Мост MQTT двунаправленного MQTT для Сетка событий Azure для проверки работы потока данных.

Экспорт конфигурации потока данных

Чтобы экспортировать конфигурацию потока данных, можно использовать портал операций или экспортировать настраиваемый ресурс потока данных.

Портал

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

Kubernetes

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