Streamen von Apache Pulsar

Wichtig

Dieses Feature befindet sich in der Public Preview.

In Databricks Runtime 14.1 und höher können Sie strukturiertes Streaming verwenden, um Daten von Apache Pulsar in Azure Databricks zu streamen.

Strukturiertes Streaming bietet eine Semantik für die einmalige Verarbeitung von Daten, die aus Pulsar-Quellen gelesen werden.

Syntaxbeispiel

Im Folgenden finden Sie ein grundlegendes Beispiel für die Verwendung von strukturiertem Streaming zum Lesen von 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()

Um aus Pulsar-Themen zu lesen, müssen Sie service.url und eine der folgenden Optionen angeben:

• topic
• topics
• topicsPattern

Eine vollständige Liste der Optionen finden Sie unter Konfigurieren von Optionen für Pulsar-Streaming-Lesevorgänge.

Authentifizieren bei Pulsar

Azure Databricks unterstützt die Truststore- und Keystore-Authentifizierung bei Pulsar. Databricks empfiehlt, geheime Schlüssel zum Speichern von Konfigurationsdetails zu verwenden.

Die vollständige Liste der Authentifizierungsoptionen finden Sie unter "Authentifizierung".

Example

Das folgende Beispiel veranschaulicht die Konfiguration der Authentifizierungsoptionen:

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

Wenn Sie von Pulsar lesen, hängt das Schema der Zeilen von den Schemas der Themen der Quelle ab.

• Bei Themen mit Avro- oder JSON-Schema bleiben Feldnamen und Feldtypen im resultierenden Spark DataFrame erhalten.
• Bei Themen ohne Schema oder mit einem einfachen Datentyp in Pulsar werden die Nutzdaten in eine value-Spalte geladen.
• Wenn Sie den Datenstrom so konfigurieren, dass mehrere Themen mit unterschiedlichen Schemas gelesen werden, legen Sie fest allowDifferentTopicSchemas , dass der rohe Inhalt in eine value Spalte geladen wird.

Pulsar-Datensätze verfügen über die folgenden Metadatenfelder:

Spalte	Typ
__key	binary
__topic	string
__messageId	binary
__publishTime	timestamp
__eventTime	timestamp
__messageProperties	map<String, String>

Optionen für das Pulsar-Streaming-Lesen konfigurieren

Die vollständige Liste der Optionen finden Sie unter Pulsar.

Start-Offsets als JSON erstellen

Informationen zur Verwendung einer benutzerdefinierten Nachrichten-ID, die einen Offset als JSON mit der startingOffsets Option angibt, finden Sie im folgenden Beispiel:

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