Leggere e scrivere i protocol buffer

Protocol Buffers (protobuf) è un formato di serializzazione binaria indipendente dal linguaggio sviluppato da Google. Azure Databricks gli utenti lo riscontrano più comunemente durante l'elaborazione di record con codifica binaria da sistemi di streaming di eventi come Apache Kafka. Azure Databricks supporta la lettura e la scrittura di dati protobuf con Apache Spark tramite le funzioni from_protobuf e to_protobuf, che convertono i dati tra protobuf binario e i tipi struct di Spark SQL, sia per i carichi di lavoro in streaming sia per quelli batch.

Prerequisiti

Le funzioni Protobuf richiedono Databricks Runtime 12.2 LTS e versioni successive.

Sintassi della funzione

Utilizzare from_protobuf per eseguire il cast di una colonna binaria in una struttura, e to_protobuf per eseguire il cast di una colonna struttura in binario. È necessario specificare un file descrittore identificato dall'argomento descFilePath o un registro schemi specificato con l'argomento options . Per un elenco completo delle opzioni, vedere Protobuf.

Pitone

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)

Linguaggio di programmazione 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

Passare le opzioni a from_protobuf e to_protobuf usando l'argomento options . Per un elenco completo delle opzioni supportate, vedere Protobuf.

Opzioni di Schema Registry

Le opzioni seguenti sono specifiche per l'utilizzo del Registro di sistema dello schema e non sono trattate nel riferimento alle opzioni generali.

Usage

Gli esempi seguenti usano il dataset Wanderbricks per mostrare la serializzazione di struct di Apache Spark in protobuf binario con to_protobuf() e la deserializzazione di record protobuf binari con from_protobuf().

Usare protobuf con Registro degli Schemi Confluent

Azure Databricks supporta l'uso del Registro schemi Confluent per definire Protobuf.

Pitone

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)

Linguaggio di programmazione 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()

eseguire l'autenticazione in un registro dello schema confluente esterno

Per eseguire l'autenticazione in un Registro schemi Confluent esterno, aggiornare le opzioni del Registro di sistema dello schema in modo da includere le credenziali di autenticazione e le chiavi API.

Pitone

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"
  }

Linguaggio di programmazione 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"
)

Usare file truststore e keystore dentro i volumi di Unity Catalog

In Databricks Runtime 14.3 LTS e versioni successive è possibile usare i file truststore e keystore nei volumi del catalogo Unity per eseguire l'autenticazione in un Registro schemi Confluent. Aggiorna le tue opzioni del registro dello schema secondo l'esempio seguente:

Pitone

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>"
  }

Linguaggio di programmazione 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>"
)

Usare Protobuf con un file descrittore

È anche possibile fare riferimento a un file di descrittore protobuf disponibile per il cluster di calcolo. Assicurati di avere le autorizzazioni appropriate per leggere il file, a seconda della sua posizione.

Pitone

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)

Linguaggio di programmazione 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()

Risorse aggiuntive

• Leggere e scrivere dati Avro in streaming: se il carico di lavoro in streaming usa la serializzazione Avro anziché Protobuf, consulta le funzioni di streaming Avro per le funzioni equivalenti a from_avro e to_avro.