Diffuser en continu à partir d’Apache Pulsar

Important

Cette fonctionnalité est disponible en préversion publique.

Dans Databricks Runtime 14.1 et versions ultérieures, vous pouvez utiliser Structured Streaming pour diffuser en continu des données à partir d’Apache Pulsar sur Azure Databricks.

Structured Streaming fournit une sémantique de traitement exactement une fois pour les données lues à partir de sources Pulsar.

Exemple de syntaxe

Voici un exemple de base d’utilisation de Structured Streaming pour lire à partir de 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()

Pour lire à partir de topics Pulsar, vous devez fournir un service.url et l’une des options suivantes :

• topic
• topics
• topicsPattern

Pour obtenir la liste complète des options, consultez Configurer les options de lecture en continu Pulsar.

S’authentifier auprès de Pulsar

Azure Databricks prend en charge l’authentification truststore et keystore sur Pulsar. Databricks vous recommande d’utiliser des secrets pour stocker les détails de la configuration.

Pour obtenir la liste complète des options d’authentification, consultez Authentification.

Example

L’exemple suivant illustre la configuration des options d’authentification :

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

Schéma Pulsar

Lorsque vous lisez à partir de Pulsar, le schéma des lignes dépend des schémas des rubriques de la source.

• Pour les rubriques avec un schéma Avro ou JSON, les noms de champs et les types de champs sont conservés dans le DataFrame Spark résultant.
• Pour les rubriques sans schéma ou avec un type de données simple dans Pulsar, la charge utile est chargée dans une colonne value.
• Si vous configurez le flux pour lire plusieurs rubriques avec différents schémas, définissez allowDifferentTopicSchemas pour charger le contenu brut dans une value colonne.

Les enregistrements Pulsar ont les champs de métadonnées suivants :

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

Configurer les options de lecture en continu de Pulsar

Pour obtenir la liste complète des options, consultez Pulsar.

Construire les offsets de départ JSON

Pour utiliser un ID de message personnalisé qui spécifie un décalage, en tant que JSON, avec l’option startingOffsets , consultez l’exemple suivant :

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