Use Apache Flink with Azure Event Hubs for Apache Kafka

  • Article

This tutorial shows you how to connect Apache Flink to an event hub without changing your protocol clients or running your own clusters. For more information on Event Hubs' support for the Apache Kafka consumer protocol, see Event Hubs for Apache Kafka.

In this tutorial, you learn how to:

  • Create an Event Hubs namespace
  • Clone the example project
  • Run Flink producer
  • Run Flink consumer

Note

This sample is available on GitHub

Prerequisites

To complete this tutorial, make sure you have the following prerequisites:

  • Read through the Event Hubs for Apache Kafka article.
  • An Azure subscription. If you do not have one, create a trial subscription before you begin.
  • Java Development Kit (JDK) 1.7+
    • On Ubuntu, run apt-get install default-jdk to install the JDK.
    • Be sure to set the JAVA_HOME environment variable to point to the folder where the JDK is installed.
  • Download and install a Maven binary archive
    • On Ubuntu, you can run apt-get install maven to install Maven.
  • Git
    • On Ubuntu, you can run sudo apt-get install git to install Git.

Create an Event Hubs namespace

An Event Hubs namespace is required to send or receive from any Event Hubs service. See Creating an event hub for instructions to create a namespace and an event hub. Make sure to copy the Event Hubs connection string for later use.

Clone the example project

Now that you have the Event Hubs connection string, clone the Azure Event Hubs for Kafka repository and navigate to the flink subfolder:

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

Using the provided Flink producer example, send messages to the Event Hubs service.

Provide an Event Hubs Kafka endpoint

producer.config

Update the bootstrap.servers and sasl.jaas.config values in producer/src/main/resources/producer.config to direct the producer to the Event Hubs Kafka endpoint with the correct authentication.

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

Important

Replace {YOUR.EVENTHUBS.CONNECTION.STRING} with the connection string for your Event Hubs namespace. For instructions on getting the connection string, see Get an Event Hubs connection string. Here's an example configuration: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.chinacloudapi.cn/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Run producer from the command line

To run the producer from the command line, generate the JAR and then run from within Maven (or generate the JAR using Maven, then run in Java by adding the necessary Kafka JAR(s) to the classpath):

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

The producer will now begin sending events to the event hub at topic test and printing the events to stdout.

Using the provided consumer example, receive messages from the event hub.

Provide an Event Hubs Kafka endpoint

consumer.config

Update the bootstrap.servers and sasl.jaas.config values in consumer/src/main/resources/consumer.config to direct the consumer to the Event Hubs Kafka endpoint with the correct authentication.

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

Important

Replace {YOUR.EVENTHUBS.CONNECTION.STRING} with the connection string for your Event Hubs namespace. For instructions on getting the connection string, see Get an Event Hubs connection string. Here's an example configuration: sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="Endpoint=sb://mynamespace.servicebus.chinacloudapi.cn/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXXXXXXXXXXXXXX";

Run consumer from the command line

To run the consumer from the command line, generate the JAR and then run from within Maven (or generate the JAR using Maven, then run in Java by adding the necessary Kafka JAR(s) to the classpath):

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

If the event hub has events (for example, if your producer is also running), then the consumer now begins receiving events from the topic test.

Check out Flink's Kafka Connector Guide for more detailed information about connecting Flink to Kafka.

Next steps

To learn more about Event Hubs for Kafka, see the following articles: