Общие сведения о схемах сообщений

Мақала
11/02/2024

Реестр схем, предоставляемый предварительной версией реестра устройств Azure, — это синхронизированный репозиторий в облаке и на границе. Реестр схем хранит определения сообщений, поступающих из пограничных ресурсов, а затем предоставляет API для доступа к этим схемам на границе.

Соединитель для OPC UA может создавать схемы сообщений и добавлять их в реестр схем или клиенты могут отправлять схемы в пользовательский веб-интерфейс операций или с помощью шаблонов ARM/Bicep.

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

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

Определения схемы сообщений

Реестр схем ожидает следующие обязательные поля в схеме сообщения:

Обязательное поле	Определение
`$schema`	`http://json-schema.org/draft-07/schema#` или `Delta/1.0`. В потоках данных схемы JSON используются для исходных конечных точек и разностных схем используются для конечных точек назначения.
`type`	`Object`
`properties`	Определение сообщения.

Примеры схем

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

JSON:

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "name": "foobarbaz",
  "description": "A representation of an event",
  "type": "object",
  "required": [ "dtstart", "summary" ],
  "properties": {
    "summary": {
      "type": "string"
    },
    "location": {
      "type": "string"
    },
    "url": {
      "type": "string"
    },
    "duration": {
      "type": "string",
      "description": "Event duration"
    }
  }
}

Дельта:

{
  "$schema": "Delta/1.0",
  "type": "object",
  "properties": {
    "type": "struct",
    "fields": [
      { "name": "asset_id", "type": "string", "nullable": false, "metadata": {} },
      { "name": "asset_name", "type": "string", "nullable": false, "metadata": {} },
      { "name": "location", "type": "string", "nullable": false, "metadata": {} },
      { "name": "manufacturer", "type": "string", "nullable": false, "metadata": {} },
      { "name": "production_date", "type": "string", "nullable": false, "metadata": {} },
      { "name": "serial_number", "type": "string", "nullable": false, "metadata": {} },
      { "name": "temperature", "type": "double", "nullable": false, "metadata": {} }
    ]
  }
}

Как потоки данных используют схемы сообщений

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

Входная схема

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

Источники ресурсов имеют предопределенную схему сообщений, созданную соединителем для OPC UA.

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

Снимок экрана: отправка схемы сообщения на портале операций.

Преобразование

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

Выходная схема

Выходные схемы связаны с назначениями потока данных, используются только для потоков данных, которые выбирают локальное хранилище, Fabric, служба хранилища Azure (ADLS 2-го поколения) или Azure Data Explorer в качестве конечной точки назначения. В настоящее время интерфейс операций Интернета вещей Azure поддерживает только выходные данные Parquet для выходных схем.

Примечание. Формат схемы Delta используется для выходных данных Parquet и Delta.

Для этих потоков данных операции применяют любые преобразования к входной схеме, а затем создает новую схему в разностном формате. При создании настраиваемого ресурса потока данных (CR) он содержит schemaRef значение, указывающее на созданную схему, хранящуюся в реестре схем.

Сведения о отправке выходной схемы см. в разделе "Отправка схемы".

Отправить схему

Входная схема может быть отправлена на портале взаимодействия с операциями, как описано в разделе схемы ввода этой статьи. Вы также можете отправить схему с помощью Azure CLI или шаблона Bicep.

Отправка схемы с помощью интерфейса командной строки

Группа команд схемы az iot ops содержит команды для создания, просмотра и управления схемами в реестре схем.

Вы можете отправить схему, ссылаясь на JSON-файл или включив схему в виде встроенного содержимого.

В следующем примере используются минимальные входные данные для создания схемы, вызываемой myschema из файла. Если номер версии не указан, версия схемы — 1.

az iot ops schema create -n myschema -g myresourcegroup --registry myregistry --format json --type message --version-content myschema.json

В следующем примере создается схема, вызываемая myschema из встроенного содержимого, и назначается номер версии.

az iot ops schema create -n myschema -g myresourcegroup --registry myregistry --format delta --type message --version-content '{\"hello\": \"world\"}' --ver 14

create После завершения команды в контейнере учетной записи хранения появится большой двоичный объект с содержимым схемы. Имя большого двоичного объекта в формате schema-namespace/schema/version.

Дополнительные параметры можно просмотреть с помощью вспомогательной команды az iot ops schema -h.

Отправка схемы с помощью шаблона Bicep

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

// Delta schema content matching OPC UA data from quickstart
// For ADLS Gen2, ADX, and Fabric destinations
var opcuaSchemaContent = '''
{
  "$schema": "Delta/1.0",
  "type": "object",
  "properties": {
    "type": "struct",
    "fields": [
      {
        "name": "temperature",
        "type": {
          "type": "struct",
          "fields": [
            {
              "name": "SourceTimestamp",
              "type": "string",
              "nullable": true,
              "metadata": {}
            },
            {
              "name": "Value",
              "type": "integer",
              "nullable": true,
              "metadata": {}
            },
            {
              "name": "StatusCode",
              "type": {
                "type": "struct",
                "fields": [
                  {
                    "name": "Code",
                    "type": "integer",
                    "nullable": true,
                    "metadata": {}
                  },
                  {
                    "name": "Symbol",
                    "type": "string",
                    "nullable": true,
                    "metadata": {}
                  }
                ]
              },
              "nullable": true,
              "metadata": {}
            }
          ]
        },
        "nullable": true,
        "metadata": {}
      },
      {
        "name": "Tag 10",
        "type": {
          "type": "struct",
          "fields": [
            {
              "name": "SourceTimestamp",
              "type": "string",
              "nullable": true,
              "metadata": {}
            },
            {
              "name": "Value",
              "type": "integer",
              "nullable": true,
              "metadata": {}
            },
            {
              "name": "StatusCode",
              "type": {
                "type": "struct",
                "fields": [
                  {
                    "name": "Code",
                    "type": "integer",
                    "nullable": true,
                    "metadata": {}
                  },
                  {
                    "name": "Symbol",
                    "type": "string",
                    "nullable": true,
                    "metadata": {}
                  }
                ]
              },
              "nullable": true,
              "metadata": {}
            }
          ]
        },
        "nullable": true,
        "metadata": {}
      }
    ]
  }
}
'''

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

// Replace placeholder values with your actual resource names
param schemaRegistryName string = '<SCHEMA_REGISTRY_NAME>'

// Pointers to existing resources from AIO deployment
resource schemaRegistry 'Microsoft.DeviceRegistry/schemaRegistries@2024-09-01-preview' existing = {
  name: schemaRegistryName
}

// Name and version of the schema
param opcuaSchemaName string = 'opcua-output-delta'
param opcuaSchemaVer string = '1'

// Define the schema resource to be created and instantiate a version
resource opcSchema 'Microsoft.DeviceRegistry/schemaRegistries/schemas@2024-09-01-preview' = {
  parent: schemaRegistry
  name: opcuaSchemaName
  properties: {
    displayName: 'OPC UA Delta Schema'
    description: 'This is a OPC UA delta Schema'
    format: 'Delta/1.0'
    schemaType: 'MessageSchema'
  }
}
resource opcuaSchemaVersion 'Microsoft.DeviceRegistry/schemaRegistries/schemas/schemaVersions@2024-09-01-preview' = {
  parent: opcSchema
  name: opcuaSchemaVer
  properties: {
    description: 'Schema version'
    schemaContent: opcuaSchemaContent
  }
}

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

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

Следующие шаги

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

Бөлісу құралы: