Kurz: proces Apache Kafka pro události Event Hubs pomocí Stream Analytics

Článek
07/12/2021
4 min ke čtení

Tento článek ukazuje, jak streamovat data do Event Hubs a zpracovávat je pomocí Azure Stream Analytics. Provede vás následujícími kroky:

Vytvořte obor názvů služby Event Hubs.
Vytvořte klienta Kafka, který odesílá zprávy do centra událostí.
Vytvořte úlohu Stream Analytics, která kopíruje data z centra událostí do úložiště objektů BLOB v Azure.

Pokud používáte koncový bod Kafka vystavený centrem událostí, nemusíte měnit klienty protokolu ani spouštět vlastní clustery. Azure Event Hubs podporuje Apache Kafka verze 1.0. a vyšší.

Požadavky

Abyste mohli absolvovat tento rychlý start, ujistěte se, že máte následující:

Předplatné Azure. Pokud ho nemáte, než začnete, vytvořte si bezplatný účet.
Java Development Kit (JDK) 1.7+.
Stáhněte a nainstalujte binární archiv Maven.
Git
účet Azure Storage. Pokud ho ještě nemáte, vytvořte ho ještě předtím, než budete pokračovat. Úloha Stream Analytics v tomto návodu ukládá výstupní data do úložiště objektů BLOB v Azure.

Vytvoření oboru názvů služby Event Hubs

Když vytvoříte obor názvů Event Hubs, je automaticky povolen koncový bod Kafka pro obor názvů. Můžete streamovat události z vašich aplikací, které používají protokol Kafka, do Center událostí. Postupujte podle podrobných pokynů v tématu vytvoření centra událostí pomocí Azure Portal k vytvoření oboru názvů Event Hubs. Pokud používáte vyhrazený cluster, přečtěte si téma Vytvoření oboru názvů a centra událostí ve vyhrazeném clusteru.

Poznámka

Event Hubs pro Kafka se na úrovni Basic nepodporuje.

Posílání zpráv pomocí Kafka v Event Hubs

Naklonujte Azure Event Hubs pro úložiště Kafka do vašeho počítače.
Přejděte do složky: azure-event-hubs-for-kafka/quickstart/java/producer .

Aktualizujte podrobnosti o konfiguraci pro producenta v src/main/resources/producer.config . Zadejte název a připojovací řetězec pro obor názvů centra událostí.

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

Přejděte na azure-event-hubs-for-kafka/quickstart/java/producer/src/main/java/ a otevřete soubor TestDataReporter. Java v editoru dle vašeho výběru.

Odkomentujte následující řádek kódu:

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

Přidejte následující řádek kódu místo kódu s komentářem:
```
            final ProducerRecord<Long, String> record = new ProducerRecord<Long, String>(TOPIC, time, "{ \"eventData\": \"Test Data " + i + "\" }");            
```
Tento kód odesílá data události ve formátu JSON . Když nakonfigurujete vstup pro Stream Analytics úlohu, jako formát vstupních dat zadáte JSON.
Spusťte producenta a stream do Event Hubs. v Windows počítači při použití příkazového řádkuNode.js přepněte do azure-event-hubs-for-kafka/quickstart/java/producer složky před spuštěním těchto příkazů.
```
mvn clean package
mvn exec:java -Dexec.mainClass="TestProducer"                                    
```

Ověřte, že centrum událostí přijímá data.

V části entity vyberte Event Hubs . Ověřte, že se zobrazuje centrum událostí s názvem test.
Ověřte, že se zobrazují zprávy přicházející do centra událostí.

Zpracování dat událostí pomocí Stream Analytics úlohy

V této části vytvoříte úlohu Azure Stream Analytics. Klient Kafka odesílá události do centra událostí. Vytvoříte úlohu Stream Analytics, která jako vstup převezme data událostí, a zapíše ji do úložiště objektů BLOB v Azure. pokud účet Azure Storage nemáte, vytvořte ho.

Dotaz v Stream Analytics úloze projde daty bez provedení jakýchkoli analýz. Můžete vytvořit dotaz, který transformuje vstupní data k vytvoření výstupních dat v jiném formátu nebo pomocí zkušeností s přehledem.

Vytvoření úlohy Stream Analytics

V Azure Portalvyberte + vytvořit prostředek .
V nabídce Azure Marketplace vyberte Analytics a vyberte Stream Analytics úloha.
Na stránce nový Stream Analytics proveďte následující akce:
1. Zadejte název úlohy.
2. Vyberte své předplatné.
3. Pro skupinu prostředků vyberte vytvořit novou a zadejte název. Můžete také použít existující skupinu prostředků.
4. Vyberte umístění pro úlohu.
5. Vyberte vytvořit a vytvořte úlohu.

Konfigurace vstupu úlohy

V oznamovací zprávě vyberte Přejít k prostředku a zobrazte stránku Stream Analytics úlohy .
V nabídce vlevo vyberte vstupy v části topologie úloh .
Vyberte Přidat vstup streamu a pak vyberte centrum událostí.
Na stránce konfigurace vstupu centra událostí proveďte následující akce:
1. Zadejte alias pro vstup.
2. Vyberte své předplatné Azure.
3. Vyberte obor názvů centra událostí , který jste vytvořili dříve.
4. Vyberte test centra událostí.
5. Vyberte Uložit.

Konfigurace výstupu úlohy

V nabídce vyberte výstupy v části topologie úloh .
Na panelu nástrojů vyberte + Přidat a vyberte úložiště objektů BLOB .
Na stránce nastavení výstupu služby Blob Storage proveďte následující akce:
1. Zadejte alias pro výstup.
2. Vyberte své předplatné Azure.
3. vyberte účet Azure Storage.
4. Zadejte název kontejneru , ve kterém jsou uložena výstupní data z Stream Analytics dotazu.
5. Vyberte Uložit.

Definovat dotaz

Po dokončení nastavení úlohy Stream Analytics pro čtení příchozího datového streamu je dalším krokem vytvoření transformace, která analyzuje data v reálném čase. Transformační dotaz definujete pomocí jazyka Stream Analytics Query Language. V tomto návodu definujete dotaz, který projde daty bez provedení jakékoli transformace.

Vyberte dotaz.
V okně dotazu nahraďte [YourOutputAlias] aliasem výstupu, který jste vytvořili dříve.
Nahraďte [YourInputAlias] vstupním aliasem, který jste vytvořili dříve.
Na panelu nástrojů vyberte Uložit.

Spuštění úlohy Stream Analytics

V nabídce vlevo vyberte Přehled .
Vyberte Spustit.
Na stránce Spustit úlohu vyberte Spustit.
Počkejte, až se stav úlohy změní z počáteční na spuštěno.

Testování scénáře

Znovu spusťte producenta Kafka , abyste mohli odesílat události do centra událostí.

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

Ověřte, že se v úložišti objektů BLOB v Azure vygenerovala výstupní data . V kontejneru se zobrazí soubor JSON s 100 řádky, které vypadají jako následující ukázkové řádky:

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

Úloha Azure Stream Analytics přijala vstupní data z centra událostí a uloží je v tomto scénáři do úložiště objektů BLOB v Azure.

Další kroky

V tomto článku jste zjistili, jak se streamovat do Event Hubs bez změny klientů protokolu nebo spouštění vlastních clusterů. Další informace o Event Hubs pro Apache Kafka najdete v tématu Apache Kafka příručka pro vývojáře pro Azure Event Hubs.