Quickstart: Valida uno schema Avro durante lo streaming di eventi con gli Event Hub .NET SDK (AMQP)

In questa guida introduttiva si apprende come inviare eventi a un hub eventi e ricevere eventi da un hub eventi con convalida dello schema usando la libreria .NET Azure.Messaging.EventHubs.

Registro schemi di Azure è una funzionalità di Hub eventi. Il Registro di sistema fornisce un repository centrale per gli schemi per le applicazioni basate su eventi e incentrate sulla messaggistica. Offre la flessibilità necessaria per consentire alle applicazioni producer e consumer di scambiare dati senza dover gestire e condividere lo schema. Fornisce inoltre un framework di governance semplice per gli schemi riutilizzabili e definisce la relazione tra gli schemi tramite un costrutto di raggruppamento (gruppi di schemi). Per altre informazioni, vedere Registro schemi di Azure in Hub eventi.

Prerequisiti

Se non si ha familiarità con Hub eventi di Azure, vedere Panoramica di Hub eventi prima di procedere con questa guida di avvio rapido.

Per completare questa guida introduttiva è necessario soddisfare i prerequisiti seguenti:

  • Se non hai una sottoscrizione di Azure, crea un account gratuito prima di iniziare.
  • Microsoft Visual Studio 2022. L'edizione Community gratuita è adatta per questo Avvio rapido.
  • L’.NET SDK 8.0 o versione successiva. I campioni di codice utilizzano istruzioni di alto livello e altre funzionalità moderne di C#, quindi usa un SDK .NET recente.

Creare un hub eventi

Per creare uno spazio dei nomi di Hub eventi e un hub eventi, seguire le istruzioni in Creare uno spazio dei nomi di Hub eventi e un hub eventi.

Per ottenere una stringa di connessione allo spazio dei nomi di Hub eventi, seguire le istruzioni riportate in Ottenere la stringa di connessione.

Nota le seguenti impostazioni da utilizzare nell'attuale avvio rapido:

  • Stringa di connessione per lo spazio dei nomi di Event Hubs
  • Nome dell'hub eventi

Creare uno schema

Per creare un gruppo di schema e uno schema, segui le istruzioni in Crea schema usando Registro degli Schema.

  1. Crea un gruppo di schema chiamato contoso-sg utilizzando il portale Schema Registre. Usare Avro come tipo di serializzazione e Nessuno per la modalità di compatibilità.

  2. In quel gruppo di schema, crea un nuovo schema Avro con il nome dello schema: Microsoft.Azure.Data.SchemaRegistry.example.Order. Usare il contenuto dello schema seguente.

    {
  "namespace": "Microsoft.Azure.Data.SchemaRegistry.example",
  "type": "record",
  "name": "Order",
  "fields": [
    {
      "name": "id",
      "type": "string"
    },
    {
      "name": "amount",
      "type": "double"
    },
    {
      "name": "description",
      "type": "string"
    }
  ]
}

Aggiungere un utente al ruolo lettore del Registro dello schema

Aggiungere l'account utente al ruolo Lettore del Registro schemi a livello di spazio dei nomi. È inoltre possibile usare il ruolo Collaboratore del Registro schemi, ma non è necessario per questo Avvio rapido.

  1. Nella pagina Namespace degli Event Hubs , seleziona Controllo Accesso (IAM) dal menu sinistro.
  2. Nella pagina Controllo di accesso (IAM) selezionare + Aggiungi>assegnazione di ruolo.
  3. Nella pagina Ruoli selezionare Lettore registro schemi e quindi selezionare Avanti.
  4. Usare il collegamento + Seleziona membri per aggiungere l'account utente al ruolo e quindi selezionare Avanti.
  5. Nella pagina Rivedi + assegna, selezionare Rivedi + assegna.

Generare eventi in hub eventi con la convalida dello schema

Creare un'applicazione console per il producer di eventi

  1. Avvia Visual Studio.

  2. Selezionare Crea un nuovo progetto.

  3. Nella finestra di dialogo Crea un nuovo progetto , completa i seguenti passaggi. Se questa finestra di dialogo non viene visualizzata, selezionare File dal menu, selezionare Nuovo e quindi selezionare Progetto.

    1. Selezionare C# come linguaggio di programmazione.

    2. Selezionare Console come tipo di applicazione.

    3. Selezionare Applicazione console nell'elenco dei risultati.

    4. Quindi seleziona Avanti.

      Screenshot che mostra la finestra di dialogo Nuovo progetto di Visual Studio.

  4. Immettere OrderProducer come nome del progetto , SRQuickStart per il nome della soluzione e quindi selezionare OK per creare il progetto.

Aggiungere il pacchetto NuGet di Hub eventi

  1. Fare clic su Strumenti>Gestione Pacchetti NuGet>Console di Gestione pacchetti.

  2. Eseguire i comandi seguenti per installare Azure.Messaging.EventHubs e altri pacchetti NuGet. Premere INVIO per eseguire l'ultimo comando.

    Install-Package Azure.Messaging.EventHubs
Install-Package Azure.Identity
Install-Package Microsoft.Azure.Data.SchemaRegistry.ApacheAvro

  3. Autenticare le applicazioni producer per connettersi ad Azure usando Visual Studio. Per altre informazioni, vedere Libreria client di Identità di Azure per .NET.

  4. Accedere ad Azure usando l'account utente membro del ruolo Schema Registry Reader a livello di spazio dei nomi. Per informazioni sui ruoli del registro degli schemi, fare riferimento a Controllo degli accessi in base al ruolo di Azure.

Generazione di codice utilizzando lo schema Avro

  1. Usare lo stesso contenuto usato per creare lo schema per creare un file denominato Order.avsc. Salvare il file nella cartella del progetto o della soluzione.
  2. Usare questo file di schema per generare codice per .NET. È possibile usare qualsiasi strumento di generazione di codice esterno, ad esempio avrogen per la generazione di codice. Ad esempio, eseguire avrogen -s .\Order.avsc . per generare codice.
  3. Dopo aver generato il codice, viene visualizzato il file denominato Order.cs nella \Microsoft\Azure\Data\SchemaRegistry\example cartella . Per lo schema Avro descritto in questo articolo, vengono generati i tipi C# nello spazio dei nomi Microsoft.Azure.Data.SchemaRegistry.example.
  4. Aggiungere il Order.cs file al OrderProducer progetto.

Scrivere codice per serializzare e inviare eventi all'hub eventi

  1. Aggiungere il codice seguente al file Program.cs. Per informazioni dettagliate, vedere i commenti sul codice. I passaggi generali nel codice sono:

    1. Creare un client producer che è possibile usare per inviare eventi a un hub eventi.
    2. Creare un client del registro degli schemi da utilizzare per serializzare e convalidare i dati in un oggetto Order.
    3. Creare un nuovo Order oggetto usando il tipo generato Order .
    4. Usare il client del registro dello schema per serializzare l'oggetto Order in EventData.
    5. Creare un batch di eventi.
    6. Aggiungere i dati dell'evento al batch di eventi.
    7. Usare il client producer per inviare il batch di eventi all'hub eventi.
    using Azure.Data.SchemaRegistry;
using Azure.Identity;
using Microsoft.Azure.Data.SchemaRegistry.ApacheAvro;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Producer;

using Microsoft.Azure.Data.SchemaRegistry.example;

// connection string to the Event Hubs namespace
const string connectionString = "EVENTHUBSNAMESPACECONNECTIONSTRING";

// name of the event hub
const string eventHubName = "EVENTHUBNAME";

// Schema Registry endpoint 
const string schemaRegistryEndpoint = "EVENTHUBSNAMESPACENAME.servicebus.windows.net";

// name of the schema group   
const string schemaGroup = "SCHEMAGROUPNAME";

// The Event Hubs client types are safe to cache and use as a singleton for the lifetime
// of the application, which is best practice when events are being published or read regularly.
EventHubProducerClient producerClient;

// Create a producer client that you can use to send events to an event hub
producerClient = new EventHubProducerClient(connectionString, eventHubName);

// Create a schema registry client that you can use to serialize and validate data.  
var schemaRegistryClient = new SchemaRegistryClient(schemaRegistryEndpoint, new DefaultAzureCredential());

// Create an Avro object serializer using the Schema Registry client object. 
var serializer = new SchemaRegistryAvroSerializer(schemaRegistryClient, schemaGroup, new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });

// Create a new order object using the generated type/class 'Order'. 
var sampleOrder = new Order { id = "1234", amount = 45.29, description = "First sample order." };
EventData eventData = (EventData)await serializer.SerializeAsync(sampleOrder, messageType: typeof(EventData));

// Create a batch of events 
using EventDataBatch eventBatch = await producerClient.CreateBatchAsync();

// Add the event data to the event batch. 
eventBatch.TryAdd(eventData);

// Send the batch of events to the event hub. 
await producerClient.SendAsync(eventBatch);
Console.WriteLine("A batch of 1 order has been published.");

  2. Sostituire i valori segnaposto seguenti con i valori reali.

    • EVENTHUBSNAMESPACECONNECTIONSTRING - stringa di connessione per lo spazio dei nomi di Event Hubs
    • EVENTHUBNAME - nome dell'hub eventi
    • EVENTHUBSNAMESPACENAME - nome dello spazio dei nomi di Hub eventi
    • SCHEMAGROUPNAME - nome del gruppo di schemi
    // connection string to the Event Hubs namespace
const string connectionString = "EVENTHUBSNAMESPACECONNECTIONSTRING";

// name of the event hub
const string eventHubName = "EVENTHUBNAME";

// Schema Registry endpoint 
const string schemaRegistryEndpoint = "EVENTHUBSNAMESPACENAME.servicebus.windows.net";

// name of the schema group   
const string schemaGroup = "SCHEMAGROUPNAME";

  3. Compilare il progetto e assicurarsi che non siano presenti errori.

  4. Eseguire il programma e attendere il messaggio di conferma.

    A batch of 1 order has been published.

  5. Nel portale di Azure è possibile verificare che l'hub eventi abbia ricevuto gli eventi. Passare alla visualizzazione Messaggi nella sezione Metriche . Aggiornare la pagina per aggiornare il grafico. Potrebbero essere necessari alcuni secondi per indicare che ha ricevuto i messaggi.

    Immagine della pagina del portale di Azure per verificare che l'hub eventi abbia ricevuto gli eventi.

Usare eventi da hub eventi con la convalida dello schema

Questa sezione mostra come scrivere un'applicazione console .NET Core che riceva eventi da un event hub e utilizza il registro dello schema per deserializzare i dati degli eventi.

Prerequisiti aggiuntivi

Creare un'applicazione consumer

  1. Nella finestra Esplora soluzioni fare clic con il pulsante destro del mouse sulla soluzione SRQuickStart , selezionare Aggiungi e selezionare Nuovo progetto.
  2. Selezionare Applicazione console e selezionare Avanti.
  3. Immettere OrderConsumer come Nome progetto e selezionare Crea.
  4. Nella finestra Esplora soluzioni fare clic con il pulsante destro del mouse su OrderConsumer e scegliere Imposta come progetto di avvio.

Aggiungere il pacchetto NuGet di Hub eventi

  1. Fare clic su Strumenti>Gestione Pacchetti NuGet>Console di Gestione pacchetti.

  2. Nella finestra Console di Gestione pacchetti verificare che OrderConsumer sia selezionato per il progetto predefinito. In caso contrario, usare l'elenco a discesa per selezionare OrderConsumer.

  3. Eseguire il comando seguente per installare i pacchetti NuGet necessari. Premere INVIO per eseguire l'ultimo comando.

    Install-Package Azure.Messaging.EventHubs
Install-Package Azure.Messaging.EventHubs.Processor
Install-Package Azure.Identity
Install-Package Microsoft.Azure.Data.SchemaRegistry.ApacheAvro
Install-Package Azure.Storage.Blobs

  4. Autenticare le applicazioni producer per connettersi ad Azure usando Visual Studio, come illustrato nella libreria client di Azure Identity per .NET.

  5. Accedi ad Azure con l'account utente che sia membro del ruolo Schema Registry Reader a livello di namespace. Per informazioni sui ruoli del registro degli schemi, fare riferimento a Controllo degli accessi in base al ruolo di Azure.

  6. Aggiungi il Order.cs file generato durante la creazione dell'app produttore al progetto OrderConsumer.

  7. Fare clic con il pulsante destro del mouse sul progetto OrderConsumer e scegliere Imposta come progetto di avvio.

Scrivere codice per ricevere gli eventi e deserializzarli usando il Registro di sistema dello schema

  1. Aggiungere il codice seguente al file Program.cs. Per informazioni dettagliate, vedere i commenti sul codice. I passaggi generali nel codice sono:

    1. Crea un client consumer che puoi usare per ricevere eventi da un hub eventi.
    2. Creare client contenitore BLOB per il contenitore BLOB in Archiviazione BLOB di Azure.
    3. Creare un client del processore di eventi e registrare i gestori di eventi e di errori.
    4. Nel gestore dell’evento creare un client del Registro schemi che è possibile usare per deserializzare i dati degli eventi in un oggetto Order.
    5. Deserializzare i dati dell'evento in un Order oggetto usando il serializzatore.
    6. Stampare le informazioni sull'ordine ricevuto.
    using Azure.Data.SchemaRegistry;
using Azure.Identity;
using Microsoft.Azure.Data.SchemaRegistry.ApacheAvro;
using Azure.Storage.Blobs;
using Azure.Messaging.EventHubs;
using Azure.Messaging.EventHubs.Consumer;
using Azure.Messaging.EventHubs.Processor;

using Microsoft.Azure.Data.SchemaRegistry.example;


// connection string to the Event Hubs namespace
const string connectionString = "EVENTHUBSNAMESPACECONNECTIONSTRING";

// name of the event hub
const string eventHubName = "EVENTHUBNAME";

// Schema Registry endpoint 
const string schemaRegistryEndpoint = "EVENTHUBSNAMESPACENAME.servicebus.windows.net";

// name of the schema group   
const string schemaGroup = "SCHEMAGROUPNAME";

// connection string for the Azure Storage account
const string blobStorageConnectionString = "AZURESTORAGECONNECTIONSTRING";

// name of the blob container that will be used as a checkpoint store
const string blobContainerName = "BLOBCONTAINERNAME";

// Create a blob container client that the event processor will use 
BlobContainerClient storageClient = new BlobContainerClient(blobStorageConnectionString, blobContainerName);

// Create an event processor client to process events in the event hub
EventProcessorClient processor = new EventProcessorClient(storageClient, EventHubConsumerClient.DefaultConsumerGroupName, connectionString, eventHubName);

// Register handlers for processing events and handling errors
processor.ProcessEventAsync += ProcessEventHandler;
processor.ProcessErrorAsync += ProcessErrorHandler;

// Start the processing
await processor.StartProcessingAsync();

// Wait for 30 seconds for the events to be processed
await Task.Delay(TimeSpan.FromSeconds(30));

// Stop the processing
await processor.StopProcessingAsync();

static async Task ProcessEventHandler(ProcessEventArgs eventArgs)
{
    // Create a schema registry client that you can use to serialize and validate data.  
    var schemaRegistryClient = new SchemaRegistryClient(schemaRegistryEndpoint, new DefaultAzureCredential());

    // Create an Avro object serializer using the Schema Registry client object. 
    var serializer = new SchemaRegistryAvroSerializer(schemaRegistryClient, schemaGroup, new SchemaRegistryAvroSerializerOptions { AutoRegisterSchemas = true });

    // Deserialized data in the received event using the schema 
    Order sampleOrder = (Order)await serializer.DeserializeAsync(eventArgs.Data, typeof(Order));

    // Print the received event
    Console.WriteLine($"Received order with ID: {sampleOrder.id}, amount: {sampleOrder.amount}, description: {sampleOrder.description}");

       await eventArgs.UpdateCheckpointAsync(eventArgs.CancellationToken);
    }

    static Task ProcessErrorHandler(ProcessErrorEventArgs eventArgs)
{
    // Write details about the error to the console window
    Console.WriteLine($"\tPartition '{eventArgs.PartitionId}': an unhandled exception was encountered. This was not expected to happen.");
    Console.WriteLine(eventArgs.Exception.Message);
    return Task.CompletedTask;
}

  2. Sostituire i valori segnaposto seguenti con i valori reali.

    • EVENTHUBSNAMESPACECONNECTIONSTRING - stringa di connessione per lo spazio dei nomi di Event Hubs
    • EVENTHUBNAME - nome dell'hub eventi
    • EVENTHUBSNAMESPACENAME - nome dello spazio dei nomi di Hub eventi
    • SCHEMAGROUPNAME - nome del gruppo di schemi
    • AZURESTORAGECONNECTIONSTRING - stringa di connessione per l'account di archiviazione di Azure
    • BLOBCONTAINERNAME - Nome del contenitore BLOB
    // connection string to the Event Hubs namespace
const string connectionString = "EVENTHUBSNAMESPACECONNECTIONSTRING";

// name of the event hub
const string eventHubName = "EVENTHUBNAME";

// Schema Registry endpoint 
const string schemaRegistryEndpoint = "EVENTHUBSNAMESPACENAME.servicebus.windows.net";

// name of the schema group   
const string schemaGroup = "SCHEMAGROUPNAME";

// Azure storage connection string
const string blobStorageConnectionString = "AZURESTORAGECONNECTIONSTRING";

// Azure blob container name
const string blobContainerName = "BLOBCONTAINERNAME";

  3. Compilare il progetto e assicurarsi che non siano presenti errori.

  4. Eseguire l'applicazione ricevente.

  5. Vedi un messaggio che indica che l'hub eventi ha ricevuto gli eventi.

    Received order with ID: 1234, amount: 45.29, description: First sample order.

    Questo evento è l'ordine che hai inviato in precedenza all'Event Hub eseguendo il programma di invio.

Esempi

Vedere Libreria client Apache Avro del Registro schemi di Azure per .NET.

Pulire le risorse

Eliminare lo spazio dei nomi di Hub eventi o eliminare il gruppo di risorse che contiene lo spazio dei nomi.

Passo successivo

Libreria client del Registro schemi di Azure per .NET

  • Last updated on