Tampons de protocole en lecture et écriture

Protocol Buffers (protobuf) est un format de sérialisation binaire indépendant du langage développé par Google. Les utilisateurs Databricks le rencontrent le plus souvent lors du traitement d'enregistrements encodés en binaire provenant de systèmes de streaming d'événements tels qu'Apache Kafka. Databricks prend en charge la lecture et l'écriture de données protobuf avec Apache Spark via les fonctions from_protobuf et to_protobuf, qui convertissent entre les types de struct protobuf binaire et Spark SQL pour les charges de travail en streaming et en batch.

Prérequis

Les fonctions Protobuf nécessitent Databricks Runtime 12.2 LTS et versions ultérieures.

Syntaxe de la fonction

Utilisez from_protobuf pour caster une colonne binaire en struct, et to_protobuf pour caster une colonne struct en binaire. Vous devez fournir un fichier descripteur identifié par l’argument descFilePath ou un registre de schémas spécifié avec l’argument options. Pour une liste complète d’options, consultez Protobuf.

Python
Scala

Python
from_protobuf(data: 'ColumnOrName', messageName: Optional[str] = None, descFilePath: Optional[str] = None, options: Optional[Dict[str, str]] = None)

to_protobuf(data: 'ColumnOrName', messageName: Optional[str] = None, descFilePath: Optional[str] = None, options: Optional[Dict[str, str]] = None)

Scala
// While using with Schema registry:
from_protobuf(data: Column, options: Map[String, String])

// Or with Protobuf descriptor file:
from_protobuf(data: Column, messageName: String, descFilePath: String, options: Map[String, String])

// While using with Schema registry:
to_protobuf(data: Column, options: Map[String, String])

// Or with Protobuf descriptor file:
to_protobuf(data: Column, messageName: String, descFilePath: String, options: Map[String, String])

Options

Passez les options à from_protobuf et to_protobuf en utilisant l'argument options. Pour une liste complète des options prises en charge, consultez Protobuf.

Options du registre de schémas

Les options suivantes sont spécifiques à l'utilisation du registre de schémas et ne sont pas couvertes dans la référence des options générales.

Option	Obligatoire	Par défaut	Description
`schema.registry.schema.evolution.mode`	Non	`"restart"`	Comment les modifications de schéma sont gérées lorsqu'un ID de schéma plus récent est détecté dans un enregistrement entrant. `"restart"` met fin à la query avec un `UnknownFieldException`; configurez les jobs pour qu'ils redémarrent en cas d'échec afin de prendre en compte les modifications. `"none"` ignore les modifications d'ID de schéma et analyse les enregistrements plus récents avec le schéma d'origine.
`confluent.schema.registry.<option>`	Non	—	Transmettez toute option de client Confluent Schema Registry en utilisant le préfixe `"confluent.schema.registry"`. Par exemple, définissez `"confluent.schema.registry.basic.auth.credentials.source"` sur `"USER_INFO"` et `"confluent.schema.registry.basic.auth.user.info"` sur `"<KEY>:<SECRET>"` pour configurer l'authentification de base.

Option	Obligatoire	Par défaut	Description
`schema.registry.schema.evolution.mode`	Non	`"restart"`	Comment les modifications de schéma sont gérées lorsqu'un ID de schéma plus récent est détecté dans un enregistrement entrant. `"restart"` met fin à la query avec un `UnknownFieldException`; configurez les jobs pour qu'ils redémarrent en cas d'échec afin de prendre en compte les modifications. `"none"` ignore les modifications d'ID de schéma et analyse les enregistrements plus récents avec le schéma d'origine.
`confluent.schema.registry.<option>`	Non	—	Transmettez toute option de client Confluent Schema Registry en utilisant le préfixe `"confluent.schema.registry"`. Par exemple, définissez `"confluent.schema.registry.basic.auth.credentials.source"` sur `"USER_INFO"` et `"confluent.schema.registry.basic.auth.user.info"` sur `"<KEY>:<SECRET>"` pour configurer l'authentification de base.

Utilisation

Les exemples suivants utilisent le dataset Wanderbricks pour démontrer la sérialisation des structures Apache Spark en protobuf binaire avec to_protobuf() et la désérialisation des enregistrements protobuf binaires avec from_protobuf().

Utiliser protobuf avec Confluent Schema Registry

Databricks prend en charge l’utilisation du Confluent Schema Registry pour définir Protobuf.

Python
Scala

Python
from pyspark.sql.protobuf.functions import to_protobuf, from_protobuf
from pyspark.sql.functions import struct

schema_registry_options = {
  "schema.registry.subject" : "app-events-value",
  "schema.registry.address" : "https://schema-registry:8081/"
}

# Serialize Wanderbricks reviews to binary Protobuf using schema registry
reviews_df = spark.read.table("samples.wanderbricks.reviews")
proto_bytes_df = reviews_df.select(
    to_protobuf(struct("review_id", "rating", "comment"), options=schema_registry_options).alias("proto_bytes")
)

# Deserialize binary Protobuf records back to a struct
reviews_restored_df = proto_bytes_df.select(
    from_protobuf("proto_bytes", options=schema_registry_options).alias("proto_event")
)
display(reviews_restored_df)

Scala
import org.apache.spark.sql.protobuf.functions._
import org.apache.spark.sql.functions.struct
import scala.collection.JavaConverters._

val schemaRegistryOptions = Map(
    "schema.registry.subject" -> "app-events-value",
    "schema.registry.address" -> "https://schema-registry:8081/"
)

// Serialize Wanderbricks reviews to binary Protobuf using schema registry
val reviewsDF = spark.read.table("samples.wanderbricks.reviews")
val protoBytesDF = reviewsDF.select(
    to_protobuf(struct($"review_id", $"rating", $"comment"), options = schemaRegistryOptions.asJava)
        .as("proto_bytes")
)

// Deserialize binary Protobuf records back to a struct
val reviewsRestoredDF = protoBytesDF.select(
    from_protobuf($"proto_bytes", options = schemaRegistryOptions.asJava)
        .as("proto_event")
)
reviewsRestoredDF.show()

S’authentifier auprès d’un registre de schémas Confluent externe

Pour vous authentifier auprès d'un Registre de schémas Confluent externe, mettez à jour vos options de registre de schémas pour inclure les informations d'identification d'authentification et les clés API.

Python
Scala

Python
schema_registry_options = {
    "schema.registry.subject" : "app-events-value",
    "schema.registry.address" : "https://remote-schema-registry-endpoint",
    "confluent.schema.registry.basic.auth.credentials.source" : "USER_INFO",
    "confluent.schema.registry.basic.auth.user.info" : "confluentApiKey:confluentApiSecret"
  }

Scala
val schemaRegistryOptions = Map(
      "schema.registry.subject" -> "app-events-value",
      "schema.registry.address" -> "https://remote-schema-registry-endpoint",
      "confluent.schema.registry.basic.auth.credentials.source" -> "USER_INFO",
      "confluent.schema.registry.basic.auth.user.info" -> "confluentApiKey:confluentApiSecret"
)

Utiliser les fichiers truststore et keystore dans les volumes Unity Catalog

Dans Databricks Runtime 14.3 LTS et versions ultérieures, vous pouvez utiliser les fichiers truststore et keystore dans les volumes Unity Catalog pour vous authentifier auprès d'un registre de schémas Confluent. Mettez à jour vos options de registre de schémas conformément à l'exemple suivant :

Python
Scala

Python
schema_registry_options = {
    "schema.registry.subject" : "app-events-value",
    "schema.registry.address" : "https://remote-schema-registry-endpoint",
    "confluent.schema.registry.ssl.truststore.location" : "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.truststore.jks",
    "confluent.schema.registry.ssl.truststore.password" : "<password>",
    "confluent.schema.registry.ssl.keystore.location" : "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.keystore.jks",
    "confluent.schema.registry.ssl.keystore.password" : "<password>",
    "confluent.schema.registry.ssl.key.password" : "<password>"
  }

Scala
val schemaRegistryOptions = Map(
      "schema.registry.subject" -> "app-events-value",
      "schema.registry.address" -> "https://remote-schema-registry-endpoint",
      "confluent.schema.registry.ssl.truststore.location" -> "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.truststore.jks",
      "confluent.schema.registry.ssl.truststore.password" -> "<password>",
      "confluent.schema.registry.ssl.keystore.location" -> "/Volumes/<catalog_name>/<schema_name>/<volume_name>/kafka.client.keystore.jks",
      "confluent.schema.registry.ssl.keystore.password" -> "<password>",
      "confluent.schema.registry.ssl.key.password" -> "<password>"
)

Utilisez Protobuf avec un fichier descripteur.

Vous pouvez également référencer un fichier descripteur protobuf disponible pour votre cluster de compute. Assurez-vous de disposer des autorisations appropriées pour lire le fichier, en fonction de son emplacement.

Python
Scala

Python
from pyspark.sql.protobuf.functions import to_protobuf, from_protobuf
from pyspark.sql.functions import struct

descriptor_file = "/path/to/proto_descriptor.desc"

# Serialize Wanderbricks reviews to binary Protobuf using a descriptor file
reviews_df = spark.read.table("samples.wanderbricks.reviews")
proto_bytes_df = reviews_df.select(
    to_protobuf(struct("review_id", "rating", "comment"), "Review", descriptor_file).alias("proto_bytes")
)

# Deserialize binary Protobuf records back to a struct
reviews_restored_df = proto_bytes_df.select(
    from_protobuf("proto_bytes", "Review", descFilePath=descriptor_file).alias("review")
)
display(reviews_restored_df)

Scala
import org.apache.spark.sql.protobuf.functions._
import org.apache.spark.sql.functions.struct

val descriptorFile = "/path/to/proto_descriptor.desc"

// Serialize Wanderbricks reviews to binary Protobuf using a descriptor file
val reviewsDF = spark.read.table("samples.wanderbricks.reviews")
val protoBytesDF = reviewsDF.select(
    to_protobuf(struct($"review_id", $"rating", $"comment"), "Review", descriptorFile).as("proto_bytes")
)

// Deserialize binary Protobuf records back to a struct
val reviewsRestoredDF = protoBytesDF.select(
    from_protobuf($"proto_bytes", "Review", descFilePath=descriptorFile).as("review")
)
reviewsRestoredDF.show()

Ressources supplémentaires

Lire et écrire des données Avro en streaming: Si votre charge de travail en streaming utilise la sérialisation Avro plutôt que Protobuf, consultez les fonctions de streaming Avro pour les fonctions équivalentes from_avro et to_avro.

Prérequis​

Syntaxe de la fonction​

Options​

Options du registre de schémas​

Utilisation​

Utiliser protobuf avec Confluent Schema Registry​

S’authentifier auprès d’un registre de schémas Confluent externe​

Utiliser les fichiers truststore et keystore dans les volumes Unity Catalog​

Utilisez Protobuf avec un fichier descripteur.​

Ressources supplémentaires​