Processamento de fluxo com Apache Kafka e Azure Databricks

Este artigo descreve como você pode usar o Apache Kafka como uma fonte ou um coletor ao executar cargas de trabalho de Streaming Estruturado no Azure Databricks.

Para mais informações sobre Kafka, consulte a documentação de Kafka.

Ler dados de Kafka

A seguir está um exemplo para uma leitura de streaming de Kafka:

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "latest")
  .load()
)

O Azure Databricks também dá suporte à semântica de leitura em lote para fontes de dados Kafka, conforme mostrado no exemplo a seguir:

df = (spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("subscribe", "<topic>")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
)

Para carregamento incremental em lote, o Databricks recomenda o uso do Kafka com Trigger.AvailableNow. Consulte A configurar o processamento em lotes incrementais.

No Databricks Runtime 13.3 LTS e superior, o Azure Databricks fornece uma função SQL para ler dados Kafka. O streaming com SQL é suportado apenas em DLT ou com tabelas de streaming em Databricks SQL. Consulte read_kafka função com valor de tabela.

Configurar o leitor de Streaming Estruturado do Kafka

O Azure Databricks fornece a palavra-chave kafka como um formato de dados para configurar conexões com o Kafka 0.10+.

A seguir estão as configurações mais comuns para Kafka:

Há várias maneiras de especificar quais tópicos assinar. Você deve fornecer apenas um destes parâmetros:

Outras configurações notáveis:

Consulte o Guia de Integração do Kafka de Streaming Estruturado para obter outras configurações opcionais.

Esquema para registros Kafka

O esquema dos registros de Kafka é:

O key e o value são sempre desserializados como matrizes de bytes com o ByteArrayDeserializer. Use operações DataFrame (como cast("string")) para desserializar explicitamente as chaves e valores.

Gravar dados em Kafka

Segue-se um exemplo de uma gravação de streaming para Kafka:

(df
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .start()
)

O Azure Databricks também dá suporte à semântica de gravação em lote para coletores de dados Kafka, conforme mostrado no exemplo a seguir:

(df
  .write
  .format("kafka")
  .option("kafka.bootstrap.servers", "<server:ip>")
  .option("topic", "<topic>")
  .save()
)

Configurar o gravador de Kafka Structured Streaming

O esquema fornecido ao DataStreamWriter interage com o coletor Kafka. Você pode usar os seguintes campos:

A seguir estão as opções comuns definidas ao escrever para Kafka:

Consulte o Guia de Integração do Kafka de Streaming Estruturado para obter outras configurações opcionais.

Recuperar métricas de Kafka

Você pode obter a média, o mínimo e o máximo do número de offsets que a consulta de streaming está atrás do último offset disponível entre todos os tópicos subscritos, utilizando as métricas avgOffsetsBehindLatest, maxOffsetsBehindLatest, e minOffsetsBehindLatest. Consulte Leitura de métricas interativamente.

Obtenha o número total estimado de bytes que o processo de consulta não consumiu dos tópicos inscritos examinando o valor de estimatedTotalBytesBehindLatest. Esta estimativa baseia-se nos lotes que foram processados nos últimos 300 segundos. O período de tempo em que a estimativa se baseia pode ser alterado definindo a opção bytesEstimateWindowLength para um valor diferente. Por exemplo, para defini-lo como 10 minutos:

df = (spark.readStream
  .format("kafka")
  .option("bytesEstimateWindowLength", "10m") # m for minutes, you can also use "600s" for 600 seconds
)

Se estiver a executar o fluxo num notebook, pode ver estas métricas na guia Dados Brutos no painel de progresso da consulta de streaming.

{
  "sources": [
    {
      "description": "KafkaV2[Subscribe[topic]]",
      "metrics": {
        "avgOffsetsBehindLatest": "4.0",
        "maxOffsetsBehindLatest": "4",
        "minOffsetsBehindLatest": "4",
        "estimatedTotalBytesBehindLatest": "80.0"
      }
    }
  ]
}

Usar SSL para conectar o Azure Databricks ao Kafka

Para habilitar conexões SSL com Kafka, siga as instruções na documentação do Confluent Criptografia e autenticação com SSL. Você pode fornecer as configurações descritas lá, prefixadas com kafka., como opções. Por exemplo, você especifica o local de armazenamento confiável na propriedade kafka.ssl.truststore.location.

A Databricks recomenda que você:

• Armazene seus certificados no armazenamento de objetos na nuvem. Você pode restringir o acesso aos certificados apenas aos clusters que podem acessar o Kafka. Consulte Governança de dados com o Catálogo Unity.
• Armazene suas senhas de certificado como segredos em um escopo secreto.

O exemplo a seguir usa locais de armazenamento de objetos e segredos do Databricks para habilitar uma conexão SSL:

df = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", ...)
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.ssl.truststore.location", <truststore-location>)
  .option("kafka.ssl.keystore.location", <keystore-location>)
  .option("kafka.ssl.keystore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<keystore-password-key-name>))
  .option("kafka.ssl.truststore.password", dbutils.secrets.get(scope=<certificate-scope-name>,key=<truststore-password-key-name>))
)

Conectar o Kafka no HDInsight ao Azure Databricks

1. Crie um cluster HDInsight Kafka.

   Consulte Conectar-se ao Kafka no HDInsight por meio de uma Rede Virtual do Azure para obter instruções.

2. Configure os corretores Kafka para anunciar o endereço correto.

   Siga as instruções em Configurar o Kafka para anunciar o IP. Se você gerencia o Kafka por conta própria nas Máquinas Virtuais do Azure, certifique-se de que a configuração advertised.listeners dos brokers esteja definida como o IP interno dos hosts.

3. Crie um cluster do Azure Databricks.

4. Emparelhe o cluster Kafka ao cluster do Azure Databricks.

   Siga as instruções em Redes virtuais entre pares.

Autenticação do principal de serviço com o Microsoft Entra ID e os Hubs de Eventos do Azure

O Azure Databricks dá suporte à autenticação de trabalhos do Spark com serviços de Hubs de Eventos. Essa autenticação é feita via OAuth com o Microsoft Entra ID.

O Azure Databricks dá suporte à autenticação de ID do Microsoft Entra com uma ID de cliente e segredo nos seguintes ambientes de computação:

• Databricks Runtime 12.2 LTS e superior em computação configurada com modo de acesso dedicado (anteriormente modo de acesso de usuário único).
• Databricks Runtime 14.3 LTS e superior em computação configurada com modo de acesso padrão (anteriormente modo de acesso compartilhado).
• Pipelines de DLT configurados sem o Unity Catalog.

O Azure Databricks não oferece suporte à autenticação de ID do Microsoft Entra com um certificado em qualquer ambiente de computação ou em pipelines DLT configurados com o Unity Catalog.

Essa autenticação não funciona na computação com modo de acesso padrão ou no Unity Catalog DLT.

Configurando o conector Kafka de streaming estruturado

Para executar a autenticação com o Microsoft Entra ID, você precisará dos seguintes valores:

• Um ID de locatário. Você pode encontrar isso na guia Serviços do Microsoft Entra ID .

• Um clientID (também conhecido como ID do aplicativo).

• Um segredo do cliente. Depois de ter isso, você deve adicioná-lo como um segredo ao seu espaço de trabalho Databricks. Para adicionar esse segredo, consulte Gerenciamento secreto.

• Um tópico do EventHubs. Pode encontrar uma lista de tópicos na secção Hubs de Eventos na secção Entidades numa página específica de Namespace de Hubs de Eventos. Para trabalhar com vários tópicos, você pode definir a função do IAM no nível dos Hubs de Eventos.

• Um servidor EventHubs. Você pode encontrar isso na página de visão geral do seu namespace específico de Hubs de Eventos:

  

Além disso, para usar o Entra ID, precisamos dizer a Kafka para usar o mecanismo SASL OAuth (SASL é um protocolo genérico, e OAuth é um tipo de "mecanismo" SASL):

• kafka.security.protocol deve ser SASL_SSL
• kafka.sasl.mechanism deve ser OAUTHBEARER
• kafka.sasl.login.callback.handler.class deve ser um nome completamente qualificado da classe Java e kafkashaded deve ser atribuído como valor ao manipulador de retorno de chamada de login da nossa classe Kafka sombreada. Veja o exemplo a seguir para a classe exata.

Exemplo

Em seguida, vejamos um exemplo em execução:

Python

# This is the only section you need to modify for auth purposes!
# ------------------------------
tenant_id = "..."
client_id = "..."
client_secret = dbutils.secrets.get("your-scope", "your-secret-name")

event_hubs_server = "..."
event_hubs_topic = "..."
# -------------------------------

sasl_config = f'kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{client_id}" clientSecret="{client_secret}" scope="https://{event_hubs_server}/.default" ssl.protocol="SSL";'

kafka_options = {
# Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers": f"{event_hubs_server}:9093",
"kafka.sasl.jaas.config": sasl_config,
"kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
"subscribe": event_hubs_topic,

# You should not need to modify these
"kafka.security.protocol": "SASL_SSL",
"kafka.sasl.mechanism": "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class": "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
}

df = spark.readStream.format("kafka").options(**kafka_options)

display(df)

Scala

// This is the only section you need to modify for auth purposes!
// -------------------------------
val tenantId = "..."
val clientId = "..."
val clientSecret = dbutils.secrets.get("your-scope", "your-secret-name")

val eventHubsServer = "..."
val eventHubsTopic = "..."
// -------------------------------

val saslConfig = s"""kafkashaded.org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="$clientId" clientSecret="$clientSecret" scope="https://$eventHubsServer/.default" ssl.protocol="SSL";"""

val kafkaOptions = Map(
// Port 9093 is the EventHubs Kafka port
"kafka.bootstrap.servers" -> s"$eventHubsServer:9093",
"kafka.sasl.jaas.config" -> saslConfig,
"kafka.sasl.oauthbearer.token.endpoint.url" -> s"https://login.microsoft.com/$tenantId/oauth2/v2.0/token",
"subscribe" -> eventHubsTopic,

// You should not need to modify these
"kafka.security.protocol" -> "SASL_SSL",
"kafka.sasl.mechanism" -> "OAUTHBEARER",
"kafka.sasl.login.callback.handler.class" -> "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
)

val scalaDF = spark.readStream
  .format("kafka")
  .options(kafkaOptions)
  .load()

display(scalaDF)

Tratamento de possíveis erros

• As opções de streaming não são suportadas.

  Se você tentar usar esse mecanismo de autenticação em um pipeline DLT configurado com o Unity Catalog, poderá receber o seguinte erro:

  

  Para resolver esse erro, use uma configuração de computação suportada. Consulte Autenticação da entidade de serviço com Microsoft Entra ID e Azure Event Hubs.

• Falha ao criar um novo KafkaAdminClient.

  Este é um erro interno que Kafka lança se qualquer uma das seguintes opções de autenticação estiver incorreta:

  • ID do cliente (também conhecido como ID do aplicativo)
  • Identificador de Inquilino
  • Servidor EventHubs

  Para resolver o erro, verifique se os valores estão corretos para essas opções.

  Além disso, poderá ver este erro se modificar as opções de configuração fornecidas por predefinição no exemplo (que lhe foi pedido para não modificar), como kafka.security.protocol.

• Não há registros sendo devolvidos

  Se você estiver tentando exibir ou processar seu DataFrame, mas não estiver obtendo resultados, verá o seguinte na interface do usuário.

  

  Essa mensagem significa que a autenticação foi bem-sucedida, mas o EventHubs não retornou nenhum dado. Algumas razões possíveis (embora não exaustivas) são:

  • Você especificou o tópico EventHubs errado.
  • A opção de configuração padrão do Kafka para startingOffsets é latest, e você ainda não está recebendo nenhum dado através do tópico. Você pode definir startingOffsetstoearliest para começar a ler dados a partir dos primeiros offsets de Kafka.