Ansluta ditt Apache Spark-program till Azure Event Hubs

Den här självstudien beskriver hur du ansluter ditt Spark-program till Event Hubs för direktuppspelning i realtid. Den här integreringen möjliggör strömning utan att du behöver ändra protokollklienter eller köra dina egna Kafka- eller Zookeeper-kluster. Den här handledningen kräver Apache Spark v2.4+ och Apache Kafka v2.0+.

I den här tutorialen lär du dig följande:

Förutsättningar

Innan du påbörjar den här självstudien kontrollerar du att du har:

• Azure-prenumeration. Om du inte har en skapar du ett kostnadsfritt konto.
• Apache Spark v2.4
• Apache Kafka v2.0
• Git

Skapa ett Event Hubs-namnområde

En Event Hubs-namnrymd krävs för att skicka och ta emot från Event Hubs-tjänster. Se Skapa en händelsehubb för instruktioner för att skapa ett namnområde och en händelsehubb. Hämta Event Hubs-anslutningssträngen och fullständigt domännamn (FQDN) för senare användning. Anvisningar finns i avsnittet om att hämta en Event Hubs-anslutningssträng.

Klona exempelprojektet

Klona Azure Event Hubs-lagringsplatsen och gå till undermappen tutorials/spark :

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

Läs från Event Hubs för Kafka

Med några konfigurationsändringar kan du börja läsa från Event Hubs för Kafka. Uppdatera BOOTSTRAP_SERVERS och EH_SASL med information från ditt namnområde och du kan börja strömma med Event Hubs på samma sätt som med Kafka. Den fullständiga exempelkoden finns i filen sparkConsumer.scala på GitHub.

//Read from your Event Hub!
val df = spark.readStream
    .format("kafka")
    .option("subscribe", TOPIC)
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.jaas.config", EH_SASL)
    .option("kafka.request.timeout.ms", "60000")
    .option("kafka.session.timeout.ms", "30000")
    .option("kafka.group.id", GROUP_ID)
    .option("failOnDataLoss", "true")
    .load()

//Use dataframe like normal (in this example, write to console)
val df_write = df.writeStream
    .outputMode("append")
    .format("console")
    .start()

Om du får ett fel som liknar följande fel lägger du till .option("spark.streaming.kafka.allowNonConsecutiveOffsets", "true") i anropet spark.readStream och försöker igen.

IllegalArgumentException: requirement failed: Got wrong record for <spark job name> even after seeking to offset 4216 got offset 4217 instead. If this is a compacted topic, consider enabling spark.streaming.kafka.allowNonConsecutiveOffsets 

Skriv till Event Hubs för Kafka

Du kan också skriva till Event Hubs på samma sätt som du skriver till Kafka. Glöm inte att uppdatera konfigurationen för att ändra BOOTSTRAP_SERVERS och EH_SASL med information från Event Hubs-namnområdet. Den fullständiga exempelkoden finns i filen sparkProducer.scala på GitHub.

df = /**Dataframe**/

//Write to your Event Hub!
df.writeStream
    .format("kafka")
    .option("topic", TOPIC)
    .option("kafka.bootstrap.servers", BOOTSTRAP_SERVERS)
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.jaas.config", EH_SASL)
    .option("checkpointLocation", "./checkpoint")
    .start()

Nästa steg

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

• Spegla en Kafkamäklare i en händelsehub
• Ansluta Apache Flink till en händelsehubb
• Integrera Kafka Connect med en händelsehubb
• Utforska exempel på vår GitHub
• Ansluta Akka Streams till en händelsehubb
• Apache Kafka-utvecklarguide för Azure Event Hubs