Strömma från Apache Pulsar

I Databricks Runtime 14.1 och senare kan du använda Structured Streaming för att strömma data från Apache Pulsar på Azure Databricks.

Structured Streaming ger semantik för exakt en gångs bearbetning för data som läses från Pulsar-källor.

Syntaxexempel

Följande är ett grundläggande exempel på hur du använder Structured Streaming för att läsa från 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()

Om du vill läsa från Pulsar-ämnen måste du ange ett service.url och något av följande alternativ:

• topic
• topics
• topicsPattern

En fullständig lista över alternativ finns i Konfigurera alternativ för Pulsar-strömning läs.

Autentisera till Pulsar

Azure Databricks stöder autentisering mot Pulsar med truststore och keystore. Databricks rekommenderar att du använder hemligheter för att lagra konfigurationsinformation.

En fullständig lista över autentiseringsalternativ finns i Autentisering.

Example

I följande exempel visas hur du konfigurerar autentiseringsalternativ:

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()

Pulsar-schema

När du läser från Pulsar beror schemat för rader på scheman för källans ämnen.

• För ämnen med Avro- eller JSON-schema bevaras fältnamn och fälttyper i den resulterande Spark DataFrame.
• För ämnen utan schema eller med en enkel datatyp i Pulsar läses nyttolasten in i kolumn value.
• Om du konfigurerar dataströmmen så att den läser flera ämnen med olika scheman anger du allowDifferentTopicSchemas att råinnehållet ska läsas in i en value kolumn.

Pulsar-poster har följande metadatafält:

Konfigurera alternativ för strömningsläsning från Pulsar

Den fullständiga listan över alternativ finns i Pulsar.

Skapa JSON för startoffsetar

Om du vill använda ett anpassat meddelande-ID som anger en förskjutning, som JSON, med startingOffsets alternativet kan du se följande exempel:

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()