Självstudie: Bearbeta Apache Kafka för Event Hubs-händelser med Stream Analytics

Artikel
06/01/2023

Den här artikeln visar hur du strömmar data till Event Hubs och bearbetar dem med Azure Stream Analytics. Den vägleder dig genom följande steg:

Skapa ett Event Hubs-namnområde.
Skapa en Kafka-klient som skickar meddelanden till händelsehubben.
Skapa ett Stream Analytics-jobb som kopierar data från händelsehubben till en Azure Blob Storage.

Du behöver inte ändra dina protokollklienter eller köra dina egna kluster när du använder Kafka-slutpunkten som exponeras av en händelsehubb. Azure Event Hubs stöder Apache Kafka version 1.0 och senare.

Förutsättningar

För att kunna slutföra den här snabbstarten behöver du följande:

En Azure-prenumeration. Om du inte har ett konto kan du skapa ett kostnadsfritt konto innan du börjar.
Java Development Kit (JDK) 1.7+.
Ladda ned och installera ett Maven-binärarkiv.
Git
Ett Azure Storage-konto. Om du inte har en skapar du en innan du fortsätter. Stream Analytics-jobbet i den här genomgången lagrar utdata i en Azure Blob Storage.

Skapa ett Event Hubs-namnområde

När du skapar ett Event Hubs-namnområde aktiveras Kafka-slutpunkten för namnområdet automatiskt. Du kan strömma händelser från dina program som använder Kafka-protokollet till händelsehubbar. Följ stegvisa instruktioner i Skapa en händelsehubb med Azure Portal för att skapa ett Event Hubs-namnområde. Om du använder ett dedikerat kluster läser du Skapa ett namnområde och en händelsehubb i ett dedikerat kluster.

Anteckning

Event Hubs för Kafka stöds inte på basic-nivån .

Skicka meddelanden med Kafka i Event Hubs

Klona Azure Event Hubs för Kafka-lagringsplatsen till datorn.
Navigera till mappen: azure-event-hubs-for-kafka/quickstart/java/producer.

Uppdatera konfigurationsinformationen för producenten i src/main/resources/producer.config. Ange namn och anslutningssträng för händelsehubbens namnområde.

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

Gå till azure-event-hubs-for-kafka/quickstart/java/producer/src/main/java/och öppna filen TestDataReporter.java i valfri redigerare.

Kommentera ut följande kodrad:

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

Lägg till följande kodrad i stället för den kommenterade koden:
```
            final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "{ \"eventData\": \"Test Data " + i + "\" }");            
```
Den här koden skickar händelsedata i JSON-format . När du konfigurerar indata för ett Stream Analytics-jobb anger du JSON som format för indata.
Kör producenten och strömma till Event Hubs. När du använder en Node.js kommandotolk på en Windows-dator växlar du till azure-event-hubs-for-kafka/quickstart/java/producer mappen innan du kör dessa kommandon.
```
mvn clean package
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```

Kontrollera att händelsehubben tar emot data

Välj Event Hubs under ENTiteter. Bekräfta att du ser en händelsehubb med namnet test.
Bekräfta att du ser meddelanden som kommer in till händelsehubben.

Bearbeta händelsedata med ett Stream Analytics-jobb

I det här avsnittet skapar du ett Azure Stream Analytics-jobb. Kafka-klienten skickar händelser till händelsehubben. Du skapar ett Stream Analytics-jobb som tar händelsedata som indata och matar ut dem till en Azure Blob Storage. Om du inte har något Azure Storage-konto skapar du ett.

Frågan i Stream Analytics-jobbet skickar data utan att utföra någon analys. Du kan skapa en fråga som transformerar indata för att skapa utdata i ett annat format eller med insikter.

Skapa ett Stream Analytics-jobb

Välj + Skapa en resurs i Azure Portal.
Välj Analys på Azure Marketplace-menyn och välj Stream Analytics-jobb.
Gör följande på sidan Ny Stream Analytics :
1. Ange ett namn för jobbet.
2. Välj din prenumeration.
3. Välj Skapa ny för resursgruppen och ange namnet. Du kan också använda en befintlig resursgrupp.
4. Välj en plats för jobbet.
5. Välj Skapa för att skapa jobbet.

Konfigurera jobbindata

I aviseringsmeddelandet väljer du Gå till resurs för att se Stream Analytics-jobbsidan .
Välj Indata i avsnittet JOBBTOPOLOGI på den vänstra menyn.
Välj Lägg till strömindata och välj sedan Händelsehubb.
Gör följande på konfigurationssidan för Event Hub-indata :
1. Ange ett alias för indata.
2. Välj din Azure-prenumeration.
3. Välj händelsehubbens namnområde som du skapade tidigare.
4. Välj test för händelsehubben.
5. Välj Spara.

Konfigurera jobbutdata

Välj Utdata i avsnittet JOBBTOPOLOGI på menyn.
Välj + Lägg till i verktygsfältet och välj Blob Storage
Gör följande på sidan Inställningar för Blob Storage-utdata:
1. Ange ett alias för utdata.
2. Välj din Azure-prenumeration.
3. Välj ditt Azure Storage-konto.
4. Ange ett namn för containern som lagrar utdata från Stream Analytics-frågan.
5. Välj Spara.

Definiera en fråga

När du har en Stream Analytics-jobbkonfiguration för att läsa en inkommande dataström är nästa steg att skapa en omvandling som analyserar data i realtid. Du definierar transformationsfrågan med Stream Analytics-frågespråket. I den här genomgången definierar du en fråga som skickar data utan att utföra någon transformering.

Välj Fråga.
I frågefönstret ersätter du [YourOutputAlias] med det utdataalias som du skapade tidigare.
Ersätt [YourInputAlias] med det indataalias som du skapade tidigare.
Välj Spara i verktygsfältet.

Köra Stream Analytics-jobbet

Välj Översikt på den vänstra menyn.
Välj Starta.
På sidan Starta jobb väljer du Start.
Vänta tills jobbets status ändras från Start till körning.

Testa scenariot

Kör Kafka-producenten igen för att skicka händelser till händelsehubben.

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

Kontrollera att utdata genereras i Azure Blob Storage. Du ser en JSON-fil i containern med 100 rader som ser ut som följande exempelrader:

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

Azure Stream Analytics-jobbet tog emot indata från händelsehubben och lagrade dem i Azure Blob Storage i det här scenariot.

Nästa steg

I den här artikeln har du lärt dig att strömma till Event Hubs utan att ändra dina protokollklienter eller köra dina egna kluster. Mer information om Event Hubs för Apache Kafka finns i Utvecklarguide för Apache Kafka för Azure Event Hubs.