Opções

Esta página descreve as opções de configuração para ler e gravar no Apache Kafka usando o Streaming Estruturado no Azure Databricks.

O conector kafka do Azure Databricks é criado com base no conector kafka do Apache Spark e dá suporte a todas as opções de configuração padrão do Kafka. Qualquer opção prefixada é kafka. passada diretamente para o cliente Kafka subjacente. Por exemplo, .option("kafka.max.poll.records", "500") define a propriedade do consumidor kafka max.poll.records . Consulte a documentação de configuração do Kafka para obter a lista completa de propriedades do Kafka disponíveis.

Para obter uma lista completa das opções de origem e coletor de Streaming Estruturado, consulte Kafka e o Guia de Integração de Streaming Estruturado + Kafka.

Opções obrigatórias

Para obter detalhes sobre as opções necessárias, consulte Kafka.

A seguinte opção é necessária para leitura e gravação:

Chave Descrição
kafka.bootstrap.servers Uma lista separada por vírgulas de endereços host:port para agentes kafka. Define a propriedade do bootstrap.servers cliente Kafka.
Se você descobrir que não há dados do Kafka, verifique se há endereços incorretos nessa lista de endereços do agente. Se a lista de endereços do agente estiver incorreta, talvez não haja erros. Os clientes kafka pressupõem que os agentes estarão disponíveis eventualmente e tentarão novamente para sempre quando receberem erros de rede.

Para leituras do Kafka, você também deve especificar exatamente uma das seguintes opções para identificar quais tópicos consumir:

  • subscribe
  • subscribePattern
  • assign

Ao gravar no Kafka, opcionalmente, você pode definir a opção topic para especificar um tópico de destino para todas as linhas. Se não estiver definido, o DataFrame deverá incluir uma topic coluna.

Opções comuns de leitor

As seguintes opções são comumente usadas ao ler do Kafka:

Chave Descrição
minPartitions O número mínimo de partições a serem lidas do Kafka.
maxRecordsPerPartition O número máximo de registros por partição do Spark.
failOnDataLoss Se a consulta deve falhar quando é possível que os dados foram perdidos.
maxOffsetsPerTrigger O número máximo de deslocamentos processados por intervalo de gatilho.
startingOffsets O deslocamento do qual a consulta começa a leitura.
endingOffsets Onde interromper a leitura de consultas em lote.
groupIdPrefix O prefixo personalizado para a ID do grupo de consumidores gerado automaticamente.
kafka.group.id A ID do grupo a ser usada durante a leitura do Kafka.
Use isso com cuidado porque pode causar um comportamento inesperado. Por padrão, cada consulta gera uma ID de grupo exclusiva para leitura de dados. Isso garante que cada consulta tenha seu próprio grupo de consumidores que evite a interferência de outros consumidores e permite que cada consulta leia todas as partições de seus tópicos assinados. Em alguns cenários, como a autorização baseada em grupo do Kafka, você pode usar IDs de grupo autorizadas específicas para ler dados.
Consultas com a mesma ID de grupo podem interferir umas nas outras e somente ler dados parciais. A interferência pode ocorrer quando você executa cargas de trabalho simultâneas em lotes e streaming ou quando você inicia e reinicia consultas em rápida sucessão.
Para minimizar problemas, defina a configuração session.timeout.ms do consumidor do Kafka como muito pequena.
includeHeaders Se deve incluir cabeçalhos de mensagem kafka na saída.
bytesEstimateWindowLength A janela de tempo usada para estimar os bytes restantes por meio da estimatedTotalBytesBehindLatest métrica.

Para obter uma lista completa das opções de origem e coletor de Streaming Estruturado, consulte Kafka e o Guia de Integração de Streaming Estruturado + Kafka.

Opções comuns de gravador

As seguintes opções são comumente usadas ao gravar no Kafka:

Chave Descrição
topic Define o tópico para todas as linhas. Isso tem precedência sobre qualquer topic coluna nos dados.
includeHeaders Se deve incluir cabeçalhos Kafka na linha.

Importante

O Databricks Runtime 13.3 LTS e versões superiores inclui uma versão mais recente da biblioteca kafka-clients que permite gravações idempotentes por padrão. Se o coletor Kafka usar a versão 2.8.0 ou inferior com ACLs configuradas, mas sem IDEMPOTENT_WRITE habilitadas, as gravações falharão. Resolva isso atualizando para o Kafka 2.8.0 ou superior ou definindo .option("kafka.enable.idempotence", "false").

Para obter uma lista completa das opções de origem e coletor de Streaming Estruturado, consulte Kafka e o Guia de Integração de Streaming Estruturado + Kafka.

Opções de autenticação

O Azure Databricks dá suporte a vários métodos de autenticação para Kafka, incluindo credenciais de serviço do Catálogo do Unity, SASL/SSL e opções específicas de nuvem para MSK do AWS, Hubs de Eventos do Azure e Kafka Gerenciado do Google Cloud.

Azure Databricks recomenda que você use as credenciais de serviço do Catálogo do Unity para autenticação em serviços Kafka gerenciados pela nuvem:

Opção Descrição
databricks.serviceCredential O nome de uma credencial de serviço do Catálogo do Unity para autenticação em serviços Kafka gerenciados pela nuvem (MSK do AWS, Hubs de Eventos do Azure ou Kafka Gerenciado pelo Google Cloud). Disponível no Databricks Runtime 16.1 e superior.
databricks.serviceCredential.scope O escopo OAuth para a credencial de serviço. Defina isso somente se o Azure Databricks não puder inferir automaticamente o escopo do serviço Kafka.

Ao usar uma credencial de serviço do Catálogo do Unity, você não precisa especificar opções SASL/SSL, como kafka.sasl.mechanism, kafka.sasl.jaas.configou kafka.security.protocol.

As opções comuns de SASL/SSL incluem:

Opção Descrição
kafka.security.protocol O protocolo usado para se comunicar com agentes (por exemplo, SASL_SSL, , SSL). PLAINTEXT
kafka.sasl.mechanism O mecanismo SASL (por exemplo, PLAIN, , SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER, ). AWS_MSK_IAM
kafka.sasl.jaas.config A cadeia de caracteres de configuração de logon do JAAS.
kafka.sasl.login.callback.handler.class O nome de classe totalmente qualificado de um manipulador de retorno de chamada de logon para autenticação SASL.
kafka.sasl.client.callback.handler.class O nome de classe totalmente qualificado de um manipulador de retorno de chamada do cliente para autenticação SASL.
kafka.ssl.truststore.location O local do arquivo de repositório de confiança SSL.
kafka.ssl.truststore.password A senha do arquivo de repositório de confiança SSL.
kafka.ssl.keystore.location O local do arquivo de repositório de chaves SSL.
kafka.ssl.keystore.password A senha do arquivo de repositório de chaves SSL.

Para obter instruções completas de configuração de autenticação, consulte Autenticação.

Recursos adicionais

  • Last updated on