Conectar seu aplicativo Apache Spark com Hubs de Eventos do Azure

Artigo
03/11/2023

Este tutorial orienta você sobre como conectar seu aplicativo Spark aos Hubs de Eventos para streaming em tempo real. Essa integração permite streaming sem a necessidade de alterar os clientes de protocolo ou de executar seus clusters Kafka ou Zookeeper. Este tutorial requer o Apache Spark v2.4+ e o Apache Kafka v2.0+.

Observação

Este exemplo está disponível no GitHub

Neste tutorial, você aprenderá como:

Criar um namespace de Hubs de Eventos
Clonar o projeto de exemplo
Executar o Spark
Ler dos Hubs de Eventos para Kafka
Gravar nos Hubs de Eventos para o Kafka

Pré-requisitos

Antes de começar este tutorial, verifique se você possui:

Assinatura do Azure. Se você não tiver uma, crie uma conta gratuita.
Apache Spark v2.4
Apache Kafka v2.0
Git

Observação

O adaptador Spark-Kafka foi atualizado para dar suporte ao Kafka v2.0 a partir do Spark v2.4. Em versões anteriores do Spark, o adaptador dava suporte ao Kafka v0.10 e versões posteriores, mas dependia especificamente das APIs do Kafka v0.10. Como os Hubs de Eventos para Kafka não dão suporte ao Kafka v0.10, os adaptadores Spark-Kafka de versões do Spark anteriores à v2.4 não têm suporte pelos Hubs de Eventos para ecossistemas do Kafka.

Criar um namespace dos Hubs de Eventos

É necessário um namespace do Hubs de Eventos para enviar e receber de qualquer serviço de Hub de Eventos. Para obter instruções sobre como criar um namespace e um hub de eventos, confira Criar um hub de eventos. Obtenha a cadeia de conexão dos Hubs de Eventos e o FQDN (nome de domínio totalmente qualificado) para uso posterior. Para obter instruções, confira Obter uma cadeia de conexão dos Hubs de Eventos.

Clonar o projeto de exemplo

Clone o repositório dos Hubs de Eventos do Azure e navegue até a subpasta tutorials/spark:

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

Ler dos Hubs de Eventos para Kafka

Com algumas alterações de configuração, você pode começar a ler dos Hubs de Eventos para Kafka. Atualize BOOTSTRAP_SERVERS e EH_SASL com detalhes de seu namespace e você poderá iniciar o streaming com os Hubs de Eventos como faria com o Kafka. Para ver o código de exemplo completo, consulte o arquivo sparkConsumer.scala no 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()

Se você receber um erro semelhante ao erro a seguir, adicione .option("spark.streaming.kafka.allowNonConsecutiveOffsets", "true") à chamada spark.readStream e tente novamente.

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

Gravar nos Hubs de Eventos para o Kafka

Você também pode gravar nos Hubs de Eventos da mesma forma que gravaria no Kafka. Não se esqueça de atualizar sua configuração para alterar BOOTSTRAP_SERVERS e EH_SASL com informações de seu namespace dos Hubs de Eventos. Para ver o código de exemplo completo, consulte o arquivo sparkProducer.scala no 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()

Próximas etapas

Para saber mais sobre os Hubs de Eventos e Hubs de Eventos para o Kafka, confira os artigos a seguir:

Compartilhar via