Zelfstudie: Apache Kafka voor Event Hubs-gebeurtenissen verwerken met behulp van Stream Analytics

Artikel
06/01/2023

In dit artikel wordt beschreven hoe u gegevens streamt naar Event Hubs en hoe u deze verwerkt met Azure Stream Analytics. U wordt door de volgende stappen geleid:

Maak een Event Hubs-naamruimte.
Maak een Kafka-client die berichten naar de Event Hub verzendt.
Maak een Stream Analytics-taak waarmee gegevens van de Event Hub naar een Azure Blob-opslag worden gekopieerd.

U hoeft uw protocolclients niet te wijzigen of uw eigen clusters uit te voeren wanneer u het Kafka-eindpunt gebruikt dat door een Event Hub wordt weergegeven. Azure Event Hubs ondersteunt Apache Kafka versie 1.0 en hoger.

Vereisten

Zorg ervoor dat u aan de volgende vereisten voldoet om deze snelstart uit te voeren:

Een Azure-abonnement. Als u nog geen account hebt, maakt u een gratis account voordat u begint.
Java Development Kit (JDK) 1.7+.
Download en installeer een binair Maven-archief.
Git
Een Azure Storage-account. Als u er nog geen hebt, maakt u er een voordat u verder gaat. Met de Stream Analytics-taak in dit overzicht worden de uitvoergegevens opgeslagen in een Azure Blob-opslag.

Een Event Hubs-naamruimte maken

Wanneer u een Event Hubs-naamruimte maakt, wordt het Kafka-eindpunt voor de naamruimte automatisch ingeschakeld. U kunt gebeurtenissen streamen vanuit uw toepassingen die gebruikmaken van het Kafka-protocol naar Event Hubs. Volg de stapsgewijze instructies in Een Event Hub maken met behulp van Azure Portal om een Event Hubs-naamruimte te maken. Als u een toegewezen cluster gebruikt, raadpleegt u Een naamruimte en Event Hub maken in een toegewezen cluster.

Notitie

Event Hubs voor Kafka wordt niet ondersteund in de Basic-laag .

Berichten verzenden met Kafka in Event Hubs

Kloon de Azure Event Hubs for Kafka-opslagplaats naar uw machine.
Navigeer naar de map azure-event-hubs-for-kafka/quickstart/java/producer.

Werk de configuratiegegevens voor de producer bij in src/main/resources/producer.config. Geef de naam en verbindingsreeks op voor de Event Hub-naamruimte.

bootstrap.servers={EVENT HUB NAMESPACE}.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{CONNECTION STRING for EVENT HUB NAMESPACE}";

Navigeer naar azure-event-hubs-for-kafka/quickstart/java/producer/src/main/java/ en open het bestand TestDataReporter.java in een editor naar keuze.

Voeg commentaar toe aan de volgende regel code:

            //final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "Test Data " + i);

Voeg de volgende regel code toe in plaats van de code met commentaar:
```
            final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "{ \"eventData\": \"Test Data " + i + "\" }");            
```
Deze code verzendt de gebeurtenisgegevens in JSON-indeling. Wanneer u invoer voor een Stream Analytics-taak configureert, geeft u JSON als de indeling voor de invoergegevens op.
Voer de producer uit en stream naar Event Hubs. Schakel op een Windows-computer, wanneer u gebruikmaakt van een Node.js-opdrachtprompt, over naar de map azure-event-hubs-for-kafka/quickstart/java/producer voordat u deze opdrachten uitvoert.
```
mvn clean package
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```

Controleren of Event Hub de gegevens ontvangt

Selecteer Event Hubs onder ENTITEITEN. Controleer of er een Event Hub met de naam test wordt weergegeven.
Controleer of er berichten worden weergegeven in de Event Hub.

Gebeurtenisgegevens verwerken met een Stream Analytics-taak

In dit gedeelte maakt u een Azure Stream Analytics-taak. De Kafka-client verzendt gebeurtenissen naar de Event Hub. U maakt een Stream Analytics-taak waarmee gebeurtenisgegevens worden ingevoerd als invoer en naar een Azure Blob-opslag worden uitgevoerd. Als u geen Azure Storage-account hebt, maakt u er een.

De query in de Stream Analytics-taak geeft de gegevens door zonder analyses uit te voeren. U kunt een query maken die de invoergegevens transformeert om uitvoergegevens te produceren in een andere indeling of met verworven inzichten.

Een Stream Analytics-taak maken

Selecteer Een resource maken in de Azure-portal.
Selecteer Analytics in het menu Azure Marketplace en selecteer Stream Analytics-taak.
Voer op de pagina Nieuwe Stream Analytics de volgende acties uit:
1. Voer een naam voor de taak in.
2. Selecteer uw abonnement.
3. Selecteer Nieuwe maken voor de resourcegroep en voer de naam in. U kunt ook een bestaande resourcegroep gebruiken.
4. Selecteer een locatie voor de taak.
5. Selecteer Maken om de taak te maken.

Taakinvoer configureren

Selecteer in het meldingsbericht Ga naar resource om de pagina Stream Analytics-taak te bekijken.
Selecteer Invoer in het gedeelte TAAKTOPOLOGIE in het menu aan de linkerkant.
Selecteer Stroominvoer toevoegen en selecteer vervolgens Event Hub.
Voer de volgende handelingen uit op de configuratiepagina Event Hub-invoer:
1. Geef een alias op voor de invoer.
2. Selecteer uw Azure-abonnement.
3. Selecteer de Event Hub-naamruimte die u eerder hebt gemaakt.
4. Selecteer test voor de Event Hub.
5. Selecteer Opslaan.

Taakuitvoer configureren

Selecteer Uitvoer in het gedeelte TAAKTOPOLOGIE in het menu.
Selecteer + Toevoegen op de werkbalk en selecteer Blob-opslag
Voer de volgende handelingen uit op de pagina met uitvoerinstellingen voor de Blob-opslag:
1. Geef een alias op voor de uitvoer.
2. Selecteer uw Azure-abonnement.
3. Selecteer uw Azure Storage-account.
4. Voer een naam in voor de container waarin de uitvoergegevens van de Stream Analytics-query worden opgeslagen.
5. Selecteer Opslaan.

Een query definiëren

Nadat u een Stream Analytics-taak zodanig hebt ingesteld dat een binnenkomende gegevensstroom kan worden gelezen, is de volgende stap het maken van een transformatie waarmee gegevens in real-time worden geanalyseerd. U definieert de transformatie-query met Stream Analytics Query Language. In dit overzicht definieert u een query die de gegevens doorgeeft zonder transformatie uit te voeren.

Selecteer Query.
Vervang in het queryvenster [YourOutputAlias] door de uitvoeralias die u eerder hebt gemaakt.
Vervang [YourInputAlias] door de invoeralias die u eerder hebt gemaakt.
Selecteer Opslaan op de werkbalk.

De Stream Analytics-taak uitvoeren

Selecteer Overzicht in het menu aan de linkerkant.
Selecteer Starten.
Selecteer op de pagina Taak starten de optie Starten.
Wacht tot de status van de taak is gewijzigd van Wordt gestart in Wordt uitgevoerd.

Het scenario testen

Voer de Kafka-producer opnieuw uit om gebeurtenissen naar de Event Hub te verzenden.

mvn exec:java -Dexec.mainClass="TestProducer"

Controleer of uitvoergegevens zijn gegenereerd in de Azure Blob-opslag. U ziet een JSON-bestand in de container met 100 rijen die er uitzien als de volgende voorbeeldrijen:

{"eventData":"Test Data 0","EventProcessedUtcTime":"2018-08-30T03:27:23.1592910Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 1","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}
{"eventData":"Test Data 2","EventProcessedUtcTime":"2018-08-30T03:27:23.3936511Z","PartitionId":0,"EventEnqueuedUtcTime":"2018-08-30T03:27:22.9220000Z"}

In dit scenario heeft de Azure Stream Analytics-taak invoergegevens ontvangen van de Event Hub en opgeslagen in de Azure Blob-opslag.

Volgende stappen

In dit artikel hebt u geleerd u hoe u kunt streamen naar Event Hubs zonder uw protocolclients te wijzigen of uw eigen clusters uit te voeren. Zie Ontwikkelaarshandleiding van Apache Kafka voor Azure Event Hubs voor meer informatie over Event Hubs voor Apache Kafka.