JSON-schema gebruiken met Apache Kafka-toepassingen

Artikel
05/22/2024

In deze zelfstudie wordt u begeleid bij een scenario waarin u JSON-schema's gebruikt om gebeurtenissen te serialiseren en deserialiseren met behulp van Azure Schema Registry in Event Hubs.

In dit geval gebruikt een Kafka-producenttoepassing JSON-schema dat is opgeslagen in Azure Schema Registry, om de gebeurtenis te serialiseren en te publiceren naar een Kafka-onderwerp/event hub in Azure Event Hubs. De Kafka-consument deserialisatie van de gebeurtenissen die het verbruikt van Event Hubs. Hiervoor wordt de schema-id van het gebeurtenis- en JSON-schema gebruikt, dat is opgeslagen in Azure Schema Registry. Diagram met de schemaserialisatie/deserialisatie voor Kafka-toepassingen met behulp van het JSON-schema.

Vereisten

Als u nog geen ervaring hebt met Azure Event Hubs, raadpleegt u het Event Hubs-overzicht voordat u deze quickstart uitvoert.

Voor het voltooien van deze snelstart moet aan de volgende vereisten worden voldaan:

Als u geen Azure-abonnement hebt, maakt u een gratis account voordat u begint.
Installeer de volgende onderdelen in uw ontwikkelomgeving:
- Java Development Kit (JDK) 1.7+.
- Download en installeer een binair Maven-archief.
- Git
Kloon het Azure Schema Registry voor kafka-opslagplaats .

Een Event Hub maken

Volg de instructies in de quickstart: Een Event Hubs-naamruimte en een Event Hub maken om een Event Hubs-naamruimte en een Event Hub te maken. Volg vervolgens de instructies van Get the verbindingsreeks om een verbindingsreeks op te halen in uw Event Hubs-naamruimte.

Noteer de volgende instellingen die u in de huidige quickstart gebruikt:

Verbinding maken iontekenreeks voor de Event Hubs-naamruimte
Naam van de Event Hub

Een schema maken

Volg de instructies uit Schemaregister maken om een schemagroep en een schema te maken.

Maak een schemagroep met de naam contoso-sg met behulp van de portal schemaregister. Gebruik JSON-schema als het serialisatietype.

Maak in die schemagroep een nieuw JSON-schema met schemanaam: Microsoft.Azure.Data.SchemaRegistry.example.CustomerInvoice met behulp van de volgende schema-inhoud.

{
  "$id": "https://example.com/person.schema.json",
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CustomerInvoice",
  "type": "object",
  "properties": {
    "invoiceId": {
      "type": "string"
    },
    "merchantId": {
      "type": "string"
    },
    "transactionValueUsd": {
      "type": "integer"
    },
    "userId": {
      "type": "string"
    }
  }
}

Een toepassing registreren voor toegang tot het schemaregister

U kunt De Microsoft Entra-id gebruiken om uw Kafka-producent- en consumententoepassing toegang te geven tot Azure Schema Registry-resources. Als u dit wilt inschakelen, moet u uw clienttoepassing registreren bij een Microsoft Entra-tenant vanuit Azure Portal.

Zie Uw toepassing registreren bij een Microsoft Entra-tenant om een Microsoft Entra-toepassing te registreren.example-app

tenant.id - stelt de tenant-id van de toepassing in
client.id - stelt de client-id van de toepassing in
client.secret - stelt het clientgeheim in voor verificatie

En als u beheerde identiteit gebruikt, hebt u het volgende nodig:

use.managed.identity.credential - geeft aan dat MSI-referenties moeten worden gebruikt, moeten worden gebruikt voor VM met MSI-functionaliteit
managed.identity.clientId- indien opgegeven, bouwt het MSI-referentie met opgegeven client-id managed.identity.resourceId - indien opgegeven, wordt MSI-referentie met opgegeven resource-id gebouwd

Gebruiker toevoegen aan de rol Schema Registry Reader

Voeg uw gebruikersaccount toe aan de rol schemaregisterlezer op naamruimteniveau. U kunt ook de rol Inzender voor schemaregisters gebruiken, maar dat is niet nodig voor deze quickstart.

Selecteer op de pagina Event Hubs-naamruimte toegangsbeheer (IAM) in het linkermenu.
Selecteer + Toevoegen -> Roltoewijzing toevoegen in het menu op de pagina Toegangsbeheer (IAM).
Selecteer Volgende op de pagina Toewijzingstype.
Selecteer Op de pagina Rollen de optie Schemaregisterlezer en selecteer vervolgens Volgende onder aan de pagina.
Gebruik de koppeling + Leden selecteren om de example-app toepassing die u in de vorige stap hebt gemaakt, toe te voegen aan de rol en selecteer vervolgens Volgende.
Selecteer Beoordelen en toewijzen op de pagina Beoordelen en toewijzen.

Clienttoepassingsconfiguratie van Kafka-toepassingen bijwerken

U moet de clientconfiguratie van de Kafka-producent- en consumententoepassingen bijwerken met de details van de Microsoft Entra-toepassing en met de schemaregistergegevens.

Als u de configuratie van de Kafka Producer wilt bijwerken, gaat u naar azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-producer.

Werk de configuratie van de Kafka-toepassing bij in src/main/resources/app.properties door de quickstartgids voor Kafka voor Event Hubs te volgen.
Werk de configuratiedetails voor de producent in src/main/resources/app.properties bij met behulp van de configuratie van het schemaregister en de Microsoft Entra-toepassing die u in de vorige stap als volgt hebt gemaakt:
```
schema.group=contoso-sg
schema.registry.url=https://<NAMESPACENAME>.servicebus.windows.net

 tenant.id=<>
 client.id=<>
 client.secret=<>
```
Volg dezelfde instructies en werk ook de configuratie azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-consumer bij.

Voor zowel Kafka-producenten- als consumententoepassingen wordt het volgende JSON-schema gebruikt:

{
  "$id": "https://example.com/person.schema.json",
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "CustomerInvoice",
  "type": "object",
  "properties": {
    "invoiceId": {
      "type": "string"
    },
    "merchantId": {
      "type": "string"
    },
    "transactionValueUsd": {
      "type": "integer"
    },
    "userId": {
      "type": "string"
    }
  }
}

Kafka-producent gebruiken met JSON-schemavalidatie

Als u de Kafka-producertoepassing wilt uitvoeren, gaat u naar azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-producer.

U kunt de producertoepassing uitvoeren zodat deze specifieke records of algemene records voor JSON-schema kan produceren. Voor de specifieke recordmodus moet u eerst de klassen genereren op basis van het producerschema met behulp van de volgende maven-opdracht:
```
mvn generate-sources
```

Vervolgens kunt u de producenttoepassing uitvoeren met behulp van de volgende opdrachten.

mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.producer.App"

Wanneer de producenttoepassing is uitgevoerd, wordt u gevraagd om het producerscenario te kiezen. Voor deze quickstart kunt u optie 1 kiezen: SpecificRecords produceren.
```
Enter case number:
1 - produce SpecificRecords
```

Wanneer het serialiseren en publiceren van gegevens is geslaagd, ziet u de volgende consolelogboeken in uw producenttoepassing:

INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 0
INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 1
INFO com.azure.schemaregistry.samples.producer.KafkaJsonSpecificRecord - Sent Order Invoice 2

Kafka-consument gebruiken met JSON-schemavalidatie

Als u de Kafka-consumententoepassing wilt uitvoeren, gaat u naar azure-schema-registry-for-kafka/tree/master/java/json/samples/kafka-consumer.

U kunt de consumententoepassing uitvoeren zodat deze JSON-schemaspecifieke records of algemene records kan gebruiken. Voor de specifieke recordmodus moet u eerst de klassen genereren op basis van het producerschema met behulp van de volgende maven-opdracht:
```
mvn generate-sources
```

Vervolgens kunt u de consumententoepassing uitvoeren met behulp van de volgende opdracht.

mvn clean package
mvn -e clean compile exec:java -Dexec.mainClass="com.azure.schemaregistry.samples.consumer.App"

Wanneer de consumententoepassing is uitgevoerd, wordt u gevraagd om het producentscenario te kiezen. Voor deze quickstart kunt u optie 1 kiezen: SpecificRecords gebruiken.
```
Enter case number:
1 - consume SpecificRecords
```

Wanneer het gegevensverbruik en deserialisatie is geslaagd, ziet u de volgende consolelogboeken in uw producenttoepassing:

INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 0, merchantId=Merchant Id 0, transactionValueUsd=0, userId=User Id 0}
INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 1, merchantId=Merchant Id 1, transactionValueUsd=1, userId=User Id 1}
INFO com.azure.schemaregistry.samples.consumer.KafkaJsonSpecificRecord - Invoice received: {invoiceId=Invoice 2, merchantId=Merchant Id 2, transactionValueUsd=2, userId=User Id 2}

Resources opschonen

Verwijder de Event Hubs-naamruimte of verwijder de resourcegroep die de naamruimte bevat.

Delen via