Informazioni sugli schemi dei messaggi

Articolo
11/20/2024

Il Registro schemi, una funzionalità fornita da Registro dispositivi di Azure, è un repository sincronizzato nel cloud e nel perimetro. Il Registro di sistema dello schema archivia le definizioni dei messaggi provenienti dagli asset perimetrali e quindi espone un'API per accedere a tali schemi al perimetro.

Il connettore per OPC UA può creare schemi di messaggio e aggiungerli al Registro schemi o i clienti possono caricare schemi nell'interfaccia utente Web dell'esperienza operativa o usando modelli ARM/Bicep.

I servizi Perimetrali usano schemi di messaggio per filtrare e trasformare i messaggi man mano che vengono instradati nello scenario perimetrale industriale.

Gli schemi sono documenti che descrivono il formato di un messaggio e il relativo contenuto per abilitare l'elaborazione e la contestualizzazione.

Definizioni dello schema dei messaggi

Il Registro di sistema dello schema prevede i campi obbligatori seguenti in uno schema di messaggio:

Campo obbligatorio	Definizione
`$schema`	`http://json-schema.org/draft-07/schema#` o `Delta/1.0`. Nei flussi di dati, gli schemi JSON vengono usati per gli endpoint di origine e gli schemi Delta vengono usati per gli endpoint di destinazione.
`type`	`Object`
`properties`	Definizione del messaggio.

Schemi di esempio

Gli schemi di esempio seguenti forniscono esempi per la definizione degli schemi dei messaggi in ogni formato.

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

Delta:

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

Generare uno schema

Per generare lo schema da un file di dati di esempio, usare l'helper di generazione dello schema.

Per un'esercitazione che usa il generatore di schemi, vedere Esercitazione: Inviare dati da un server OPC UA ad Azure Data Lake Storage Gen 2.

Come i flussi di dati usano gli schemi dei messaggi

Gli schemi dei messaggi vengono usati in tutte e tre le fasi di un flusso di dati: definizione dell'input di origine, applicazione delle trasformazioni dei dati e creazione dell'output di destinazione.

Schema di input

Ogni origine del flusso di dati può facoltativamente specificare uno schema di messaggio. Attualmente, i flussi di dati non eseguono la convalida di runtime sugli schemi dei messaggi di origine.

Le origini asset hanno uno schema di messaggio predefinito creato dal connettore per OPC UA.

Gli schemi possono essere caricati per le origini MQTT. Attualmente, Azure IoT Operations supporta JSON per gli schemi di origine, noti anche come schemi di input. Nell'esperienza operativa è possibile selezionare uno schema esistente o caricarne uno durante la definizione di un'origine MQTT:

Screenshot che mostra il caricamento di uno schema di messaggio nel portale dell'esperienza operativa.

Trasformazione

L'esperienza operativa usa lo schema di input come punto di partenza per i dati, semplificando la selezione delle trasformazioni in base al formato del messaggio di input noto.

Schema di output.

Gli schemi di output sono associati alle destinazioni del flusso di dati.

Nel portale dell'esperienza operativa è possibile configurare gli schemi di output per gli endpoint di destinazione seguenti che supportano l'output Parquet:

archiviazione locale
Infrastruttura OneLake
Archiviazione di Azure (ADLS Gen2)
Esplora dati di Azure

Nota: il formato dello schema Delta viene usato sia per l'output Parquet che per l'output Delta.

Se si usa Bicep o Kubernetes, è possibile configurare gli schemi di output usando l'output JSON per gli endpoint di destinazione MQTT e Kafka. Le destinazioni basate su MQTT e Kafka non supportano il formato Delta.

Per questi flussi di dati, l'esperienza operativa applica tutte le trasformazioni allo schema di input e quindi crea un nuovo schema in formato Delta. Quando viene creata la risorsa personalizzata del flusso di dati , include un schemaRef valore che punta allo schema generato archiviato nel Registro di sistema dello schema.

Per caricare uno schema di output, vedere Caricare lo schema.

Carica schema

Lo schema di input può essere caricato nel portale dell'esperienza operativa, come descritto nella sezione Schema di input di questo articolo. È anche possibile caricare uno schema usando l'interfaccia della riga di comando di Azure o un modello Bicep.

Caricare lo schema con l'interfaccia della riga di comando

Il gruppo di comandi az iot ops schema contiene i comandi per creare, visualizzare e gestire gli schemi nel Registro di sistema dello schema.

È possibile caricare uno schema facendo riferimento a un file JSON o includendo lo schema come contenuto inline.

Nell'esempio seguente vengono usati input minimi per creare uno schema chiamato myschema da un file. Quando non viene specificato alcun numero di versione, la versione dello schema è 1.

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

Nell'esempio seguente viene creato uno schema denominato myschema dal contenuto inline e viene assegnato un numero di versione.

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

Suggerimento

Se non si conosce il nome del Registro di sistema, usare il schema registry list comando per eseguirne una query. Ad esempio:

az iot ops schema registry list -g myresourcegroup --query "[].{Name:name}" -o tsv

Al termine del create comando, verrà visualizzato un BLOB nel contenitore dell'account di archiviazione con il contenuto dello schema. Il nome del BLOB è nel formato schema-namespace/schema/version.

È possibile visualizzare altre opzioni con il comando az iot ops schema -hhelper .

Caricare lo schema con un modello Bicep

Creare un file Bicep .bicep e aggiungervi il contenuto dello schema nella parte superiore come variabile. Questo esempio è uno schema Delta che corrisponde ai dati OPC UA della guida introduttiva.

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

Quindi, nello stesso file, appena sotto lo schema, definire la risorsa dello schema insieme ai puntatori alla risorsa del Registro di sistema dello schema esistente che è stata eseguita dalla distribuzione di Operazioni IoT di 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
  }
}

Dopo aver definito il contenuto e le risorse dello schema, è possibile distribuire il modello Bicep per creare lo schema nel Registro di sistema dello schema.

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

Passaggi successivi

Creare un flusso di dati

Condividi tramite