Fluxo do Apache Pulsar

No Databricks Runtime 14.1 e superior, você pode usar o Streaming Estruturado para transmitir dados do Apache Pulsar no Azure Databricks.

O Streaming Estruturado fornece semântica de processamento exatamente uma vez para dados lidos de fontes do Pulsar.

Exemplo de sintaxe

A seguir está um exemplo básico do uso do Structured Streaming para ler do Pulsar:

Python

query = (spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .load()
)

Scala

val query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .load()

Para ler tópicos do Pulsar, deve fornecer uma service.url e uma das seguintes opções:

• topic
• topics
• topicsPattern

Para obter uma lista completa de opções, consulte Configurar opções para a transmissão em fluxo do Pulsar.

Autenticar-se no Pulsar

O Azure Databricks suporta autenticação com truststore e keystore no Pulsar. O Databricks recomenda que use segredos para armazenar detalhes de configuração.

As opções de configuração de stream disponíveis incluem as seguintes:

• pulsar.client.authPluginClassName
• pulsar.client.authParams
• pulsar.client.useKeyStoreTls
• pulsar.client.tlsTrustStoreType
• pulsar.client.tlsTrustStorePath
• pulsar.client.tlsTrustStorePassword

Se o fluxo usar um PulsarAdmin, deve definir as seguintes opções:

• pulsar.admin.authPluginClassName
• pulsar.admin.authParams

Example

O exemplo a seguir demonstra a configuração de opções de autenticação:

Python

client_auth_params = dbutils.secrets.get(scope="pulsar", key="clientAuthParams")
client_pw = dbutils.secrets.get(scope="pulsar", key="clientPw")

# clientAuthParams is a comma-separated list of key-value pairs, such as:
# "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw"

query = (spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .option("startingOffsets", starting_offsets)
  .option("pulsar.client.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls")
  .option("pulsar.client.authParams", client_auth_params)
  .option("pulsar.client.useKeyStoreTls", "true")
  .option("pulsar.client.tlsTrustStoreType", "JKS")
  .option("pulsar.client.tlsTrustStorePath", trust_store_path)
  .option("pulsar.client.tlsTrustStorePassword", client_pw)
  .load()
)

Scala

val clientAuthParams = dbutils.secrets.get(scope = "pulsar", key = "clientAuthParams")
val clientPw = dbutils.secrets.get(scope = "pulsar", key = "clientPw")

// clientAuthParams is a comma-separated list of key-value pairs, such as:
// "keyStoreType:JKS,keyStorePath:/var/private/tls/client.keystore.jks,keyStorePassword:clientpw"

val query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topics", "topic1,topic2")
  .option("startingOffsets", startingOffsets)
  .option("pulsar.client.authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationKeyStoreTls")
  .option("pulsar.client.authParams", clientAuthParams)
  .option("pulsar.client.useKeyStoreTls", "true")
  .option("pulsar.client.tlsTrustStoreType", "JKS")
  .option("pulsar.client.tlsTrustStorePath", trustStorePath)
  .option("pulsar.client.tlsTrustStorePassword", clientPw)
  .load()

Esquema pulsar

Quando se lê do Pulsar, o esquema das linhas depende dos esquemas dos tópicos da fonte.

• Para tópicos com esquema Avro ou JSON, os nomes e tipos de campo são preservados no Spark DataFrame resultante.
• Para tópicos sem esquema ou com um tipo de dados simples no Pulsar, a carga útil é carregada numa coluna value.
• Se configurares o stream para ler vários tópicos com esquemas diferentes, define allowDifferentTopicSchemas para carregar o conteúdo bruto numa value coluna.

Os registros Pulsar têm os seguintes campos de metadados:

Configurar opções para leitura de streaming do Pulsar

Configure todas as opções seguintes com a sintaxe .option("<optionName>", "<optionValue>") para fluxos de leitura. Também pode configurar a autenticação usando .options(). Consulte Autenticar-se no Pulsar.

A tabela a seguir descreve as configurações necessárias para o Pulsar. Você deve especificar apenas uma das opções topicou topicstopicsPattern.

A tabela a seguir descreve outras opções suportadas para o Pulsar:

Você também pode especificar qualquer configuração de cliente, administrador e leitor do Pulsar usando os seguintes padrões:

Construir deslocamentos iniciais JSON

Para usar um ID de mensagem personalizado que especifique um deslocamento, como JSON, com a startingOffsets opção, veja o seguinte exemplo:

import org.apache.spark.sql.pulsar.JsonUtils
import org.apache.pulsar.client.api.MessageId
import org.apache.pulsar.client.impl.MessageIdImpl

val topic = "my-topic"
val msgId: MessageId = new MessageIdImpl(ledgerId, entryId, partitionIndex)
val startOffsets = JsonUtils.topicOffsets(Map(topic -> msgId))

query = spark.readStream
  .format("pulsar")
  .option("service.url", "pulsar://broker.example.com:6650")
  .option("topic", topic)
  .option("startingOffsets", startOffsets)
  .load()