Stream from Apache Pulsar

In Databricks Runtime 14.1 and above, you can use Structured Streaming to stream data from Apache Pulsar on Azure Databricks.

Structured Streaming provides exactly-once processing semantics for data read from Pulsar sources.

Syntax example

The following is a basic example of using Structured Streaming to read from Pulsar:

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

You must always provide a service.url and one of the following options to specify topics:

• topic
• topics
• topicsPattern

For a complete list of options, see Configure options for Pulsar streaming read.

Authenticate to Pulsar

Azure Databricks supports truststore and keystore authentication to Pulsar. Databricks recommends using secrets when storing configuration details.

You can set the following options during stream configuration:

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

If the stream uses a PulsarAdmin, also set the following:

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

The following example demonstrates configuring authentication options:

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"

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

The schema of records read from Pulsar depends on how topics have their schemas encoded.

• For topics with Avro or JSON schema, field names and field types are preserved in the resulting Spark DataFrame.
• For topics without schema or with a simple data type in Pulsar, the payload is loaded to a value column.
• If the reader is configured to read multiple topics with different schemas, set allowDifferentTopicSchemas to load the raw content to a value column.

Pulsar records have the following metadata fields:

Column	Type
__key	binary
__topic	string
__messageId	binary
__publishTime	timestamp
__eventTime	timestamp
__messageProperties	map<String, String>

Configure options for Pulsar streaming read

All options are configured as part of a Structured Streaming read using .option("<optionName>", "<optionValue>") syntax. You can also configure authentication using options. See Authenticate to Pulsar.

The following table describes required configurations for Pulsar. You must specify only one of the options topic, topics or topicsPattern.

Option	Default value	Description
service.url	none	The Pulsar serviceUrl configuration for the Pulsar service.
topic	none	A topic name string for the topic to consume.
topics	none	A comma-separated list of the topics to consume.
topicsPattern	none	A Java regex string to match on topics to consume.

The following table describes other options supported for Pulsar:

Option	Default value	Description
predefinedSubscription	none	The predefined subscription name used by the connector to track spark application progress.
subscriptionPrefix	none	A prefix used by the connector to generate a random subscription to track spark application progress.
pollTimeoutMs	120000	The timeout for reading messages from Pulsar in milliseconds.
waitingForNonExistedTopic	false	Whether the connector should wait until the desired topics are created.
failOnDataLoss	true	Controls whether to fail a query when data is lost (for example, topics are deleted, or messages are deleted because of retention policy).
allowDifferentTopicSchemas	false	If multiple topics with different schemas are read, use this parameter to turn off automatic schema-based topic value deserialization. Only the raw values are returned when this is true.
startingOffsets	latest	If latest, the reader reads the newest records after it starts running. If earliest, the reader reads from the earliest offset. The user can also specify a JSON string that specifies a specific offset.
maxBytesPerTrigger	none	A soft limit of the maximum number of bytes we want to process per microbatch. If this is specified, admin.url also needs to be specified.
admin.url	none	The Pulsar serviceHttpUrl configuration. Only needed when maxBytesPerTrigger is specified.

You can also specify any Pulsar client, admin, and reader configurations using the following patterns:

Pattern	Link to conifiguration options
pulsar.client.*	Pulsar client configuration
pulsar.admin.*	Pulsar admin configuration
pulsar.reader.*	Pulsar reader configuration

Construct starting offsets JSON

You can manually construct a message ID to specify a specific offset and pass this as a JSON to the startingOffsets option. The following code example demonstrates this syntax:

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