Čtení a zápis protokolových bufferů

Protocol Buffers (protobuf) je jazykově neutrální binární serializační formát vyvinutý Společností Google. Azure Databricks se s nimi uživatelé nejčastěji setkávají při zpracování binárních záznamů zakódovaných ze systémů streamování událostí, jako je Apache Kafka. Azure Databricks podporuje čtení a zápis dat ve formátu Protobuf v Apache Sparku prostřednictvím funkcí from_protobuf a to_protobuf, které převádějí mezi binárním formátem Protobuf a typy struktur Spark SQL pro streamingové i dávkové úlohy.

Předpoklady

Funkce Protobuf vyžadují Databricks Runtime 12.2 LTS a vyšší.

Syntaxe funkce

Slouží from_protobuf k přetypování binárního sloupce na strukturu a to_protobuf k přetypování sloupce struktury na binární. Je nutné zadat soubor popisovače identifikovaný argumentem descFilePath nebo registr schématu zadaný argumentem options . Úplný seznam možností najdete v tématu Protobuf.

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

Možnosti

Předejte možnosti do from_protobuf a to_protobuf pomocí argumentu options. Úplný seznam podporovaných možností najdete v tématu Protobuf.

Možnosti služby Schema Registry

Následující možnosti jsou specifické pro použití registru schématu a nejsou popsány v obecných referenčních informacích k možnostem.

Možnost Požaduje se Default Description
schema.registry.schema.evolution.mode Ne "restart" Způsob zpracování změn schématu při zjištění novějšího ID schématu v příchozím záznamu "restart" ukončí dotaz pomocí UnknownFieldException; nakonfigurujte úlohy tak, aby se při selhání restartovaly a načetly změny. "none" ignoruje změny id schématu a parsuje novější záznamy s původním schématem.
confluent.schema.registry.<option> Ne — Předejte libovolnou volbu klienta služby Confluent Schema Registry pomocí předpony "confluent.schema.registry". Můžete například nastavit "confluent.schema.registry.basic.auth.credentials.source""USER_INFO" a "confluent.schema.registry.basic.auth.user.info""<KEY>:<SECRET>" nakonfigurovat základní ověřování.

Usage

Následující příklady používají datovou sadu Wanderbricks k demonstraci serializace struktur Apache Spark do binárního protobufu pomocí to_protobuf() a deserializace záznamů binárního protobufu pomocí from_protobuf().

Použijte protobuf s registrem schématu Confluent

Azure Databricks podporuje definování Protobuf pomocí registru schémat Confluent .

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

Přihlaste se k externímu registru schémat Confluent

Pokud se chcete ověřit v externím registru schématu Confluent, aktualizujte možnosti registru schématu tak, aby zahrnovaly přihlašovací údaje ověřování a klíče rozhraní API.

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

Použijte důvěryhodnostní úložiště a klíčové úložiště ve svazcích katalogu Unity

Ve službě Databricks Runtime 14.3 LTS a vyšší můžete k ověření v registru schémat Confluent použít úložiště důvěryhodnosti a soubory úložiště klíčů ve svazcích katalogu Unity. Aktualizujte možnosti registru schématu podle následujícího příkladu:

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

Použijte Protobuf se souborem popisu

Můžete také odkazovat na soubor popisovače protobuf, který je dostupný pro váš výpočetní cluster. Ujistěte se, že máte správná oprávnění ke čtení souboru v závislosti na jeho umístění.

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

Dodatečné zdroje

  • Čtení a zápis streamovacích dat Avro: Pokud vaše streamovací úloha používá serializaci Avro namísto Protobufu, podívejte se na streamovací funkce Avro, kde najdete ekvivalentní funkce from_avro a to_avro.
  • Last updated on