Dela via

Använda Apache Flink med Azure Event Hubs för Apache Kafka

  • Artikel

Den här självstudien visar hur du ansluter Apache Flink till en händelsehubb utan att ändra dina protokollklienter eller köra egna kluster. Mer information om Event Hubs stöd för Apache Kafka-konsumentprotokollet finns i Event Hubs för Apache Kafka.

I den här guiden får du lära dig att:

  • Skapa ett Event Hubs-namnområde
  • Klona exempelprojektet
  • Kör Flink-producent
  • Kör Flink-konsument

Anteckning

Det här exemplet finns på GitHub

Förutsättningar

Kontrollera att du har följande förutsättningar för att slutföra den här självstudien:

  • Läs artikeln Event Hubs för Apache Kafka.
  • 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+
    • I Ubuntu kör du apt-get install default-jdk för att installera JDK-paketet.
    • Tänk på att ställa in miljövariabeln JAVA_HOME så att den pekar på den mapp där JDK-paketet är installerat.
  • Ladda ned och installera ett binärt Maven-arkiv
    • I Ubuntu kan du köra apt-get install maven för att installera Maven.
  • Git
    • I Ubuntu kan du köra sudo apt-get install git för att installera Git.

Skapa ett Event Hubs-namnområde

Ett Event Hubs-namnområde krävs för att skicka eller ta emot från en Event Hubs-tjänst. Se Skapa en händelsehubb för instruktioner för att skapa ett namnområde och en händelsehubb. Se till att kopiera Event Hubs-anslutningssträngen för senare användning.

Klona exempelprojektet

Nu när du har händelsehubbens anslutningssträng klonar du Azure Event Hubs för Kafka-lagringsplatsen och navigerar till undermappenflink:

git clone https://github.com/Azure/azure-event-hubs-for-kafka.git
cd azure-event-hubs-for-kafka/tutorials/flink

Med hjälp av det angivna Flink-producentexemplet skickar du meddelanden till Event Hubs-tjänsten.

Ange en Event Hubs Kafka-slutpunkt

producer.config

bootstrap.servers Uppdatera värdena och sasl.jaas.config i producer/src/main/resources/producer.config för att dirigera producenten till Event Hubs Kafka-slutpunkten med rätt autentisering.

bootstrap.servers={YOUR.EVENTHUBS.FQDN}:9093
client.id=FlinkExampleProducer
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="$ConnectionString" \
   password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

Viktigt

Ersätt {YOUR.EVENTHUBS.CONNECTION.STRING} med anslutningssträngen för Event Hubs-namnområdet. Anvisningar om hur du hämtar anslutningssträngen finns i Hämta en Händelsehubb-anslutningssträng. Här är en exempelkonfiguration: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Kör producent från kommandoraden

Om du vill köra producenten från kommandoraden genererar du JAR-filen och kör sedan inifrån Maven (eller genererar JAR-filen med Maven och kör sedan i Java genom att lägga till nödvändiga Kafka JAR:ar i klassökvägen):

mvn clean package
mvn exec:java -Dexec.mainClass="FlinkTestProducer"

Producenten börjar nu skicka händelser till händelsehubben i ämnet test och skriva ut händelserna till stdout.

Med hjälp av det angivna konsumentexemplet tar du emot meddelanden från händelsehubben.

Ange en Event Hubs Kafka-slutpunkt

consumer.config

bootstrap.servers Uppdatera värdena och sasl.jaas.config i consumer/src/main/resources/consumer.config för att dirigera konsumenten till Event Hubs Kafka-slutpunkten med rätt autentisering.

bootstrap.servers={YOUR.EVENTHUBS.FQDN}:9093
group.id=FlinkExampleConsumer
sasl.mechanism=PLAIN
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
   username="$ConnectionString" \
   password="{YOUR.EVENTHUBS.CONNECTION.STRING}";

Viktigt

Ersätt {YOUR.EVENTHUBS.CONNECTION.STRING} med anslutningssträngen för Event Hubs-namnområdet. Anvisningar om hur du hämtar anslutningssträngen finns i Hämta en Händelsehubb-anslutningssträng. Här är en exempelkonfiguration: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Kör konsument från kommandoraden

Om du vill köra konsumenten från kommandoraden genererar du JAR-filen och kör sedan inifrån Maven (eller genererar JAR-filen med Maven och kör sedan i Java genom att lägga till nödvändiga Kafka JAR:ar i klassökvägen):

mvn clean package
mvn exec:java -Dexec.mainClass="FlinkTestConsumer"

Om händelsehubben har händelser (till exempel om din producent också körs) börjar konsumenten nu ta emot händelser från ämnet test.

Se Flinks Kafka Connector Guide för mer detaljerad information om hur du ansluter Flink till Kafka.

Nästa steg

Mer information om Event Hubs för Kafka finns i följande artiklar: