Odczyt i zapis buforów protokołów

Bufory protokołu (protobuf) to neutralny dla języka format serializacji binarnej opracowany przez firmę Google. Azure Databricks użytkownicy najczęściej napotykają go podczas przetwarzania rekordów zakodowanych binarnie z systemów przesyłania strumieniowego zdarzeń, takich jak Apache Kafka. Azure Databricks obsługuje odczytywanie i zapisywanie danych protobuf w Apache Spark za pomocą funkcji from_protobuf i to_protobuf, które konwertują dane między binarnym formatem protobuf a typami strukturalnymi Spark SQL, zarówno w zadaniach strumieniowych, jak i wsadowych.

Wymagania wstępne

Funkcje Protobuf wymagają środowiska Databricks Runtime 12.2 LTS lub nowszego.

Składnia funkcji

Użyj from_protobuf, aby rzutować kolumnę binarną na typ struct, a to_protobuf, aby rzutować kolumnę typu struct do postaci binarnej. Musisz podać plik deskryptora zidentyfikowany przez descFilePath argument lub rejestr schematów określony z argumentem options . Aby uzyskać pełną listę opcji, zobacz 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)

Skala

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

Opcje

Przekaż opcje do from_protobuf i to_protobuf za pomocą argumentu options. Aby uzyskać pełną listę obsługiwanych opcji, zobacz Protobuf.

Opcje rejestru schematów

Poniższe opcje są specyficzne dla użycia rejestru schematów i nie są omówione w dokumentacji ogólnych opcji.

Option	Wymagane	Default	Description
`schema.registry.schema.evolution.mode`	No	`"restart"`	Sposób obsługi zmian schematu w przypadku wykrycia nowszego identyfikatora schematu w rekordzie przychodzącym. `"restart"` kończy zapytanie za pomocą polecenia `UnknownFieldException`; skonfiguruj zadania do ponownego uruchomienia po niepowodzeniu pobrania zmian. `"none"` Ignoruje zmiany identyfikatora schematu i analizuje nowsze rekordy przy użyciu oryginalnego schematu.
`confluent.schema.registry.<option>`	No	—	Przekaż dowolną opcję klienta Rejestru schematów Confluent za pomocą prefiksu `"confluent.schema.registry"`. Na przykład ustaw `"confluent.schema.registry.basic.auth.credentials.source"` na `"USER_INFO"` i `"confluent.schema.registry.basic.auth.user.info"` na `"<KEY>:<SECRET>"`, aby skonfigurować uwierzytelnianie podstawowe.

Usage

Poniższe przykłady wykorzystują zbiór danych Wanderbricks dataset do zademonstrowania serializacji struktur Apache Spark do binarnego formatu protobuf przy użyciu to_protobuf() oraz deserializacji binarnych rekordów protobuf przy użyciu from_protobuf().

Używanie narzędzia protobuf z rejestrem schematów platformy Confluent

Usługa Azure Databricks obsługuje definiowanie narzędzia Protobuf przy użyciu rejestru schematów 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)

Skala

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

Uwierzytelnij się w zewnętrznym rejestrze schematów Confluent

Aby uwierzytelnić się w zewnętrznym rejestrze schematów confluent, zaktualizuj opcje rejestru schematów, aby uwzględnić poświadczenia uwierzytelniania i klucze interfejsu 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"
  }

Skala

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

Korzystanie z plików truststore i keystore w woluminach Unity Catalog

W środowisku Databricks Runtime 14.3 LTS i nowszym można użyć plików repozytoriów zaufania i kluczy w wolumenach Unity Catalog, aby uwierzytelnić się w Rejestrze schematów Confluent. Zaktualizuj opcje rejestru schematów zgodnie z poniższym przykładem:

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

Skala

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

Używanie narzędzia Protobuf z plikiem deskryptora

Możesz również odwołać się do pliku deskryptora protobuf dostępnego dla klastra obliczeniowego. Upewnij się, że masz odpowiednie uprawnienia do odczytu pliku, w zależności od jego lokalizacji.

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)

Skala

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

Dodatkowe zasoby

Odczytywanie i zapisywanie strumieniowych danych Avro: Jeśli obciążenie przetwarzania strumieniowego używa serializacji Avro zamiast Protobuf, zobacz funkcje strumieniowe Avro, aby poznać odpowiedniki funkcji from_avro i to_avro.

Opinia

Czy ta strona była pomocna?

Last updated on 2026-06-24