Okuma ve yazma protokolü arabellekleri

Protokol Arabellekleri (protobuf), Google tarafından geliştirilen dilden bağımsız bir ikili serileştirme biçimidir. Azure Databricks kullanıcılar, Apache Kafka gibi olay akış sistemlerinden ikili kodlanmış kayıtları işlerken en yaygın olarak bununla karşılaşır. Azure Databricks, Apache Spark’ta from_protobuf ve to_protobuf işlevleri aracılığıyla protobuf verilerini okumayı ve yazmayı destekler; bu işlevler, hem akış hem de toplu iş yükleri için ikili protobuf ile Spark SQL struct türleri arasında dönüşüm yapar.

Prerequisites

Protobuf işlevleri Databricks Runtime 12.2 LTS ve üzerini gerektirir.

İşlev söz dizimi

from_protobuf, ikili bir sütunu bir yapıya dönüştürmek için, to_protobuf ise bir yapı sütununu ikiliye dönüştürmek için kullanılır. descFilePath argümanıyla tanımlanan bir tanımlayıcı dosyası veya options argümanıyla belirtilen bir şema kayıt deposu sağlamanız gerekir. Seçeneklerin tam listesi için bkz. Protobuf.

Piton

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 programlama dili

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

options bağımsız değişkenini kullanarak seçenekleri from_protobuf ve to_protobuf öğelerine iletin. Desteklenen seçeneklerin tam listesi için bkz. Protobuf.

Şema Kayıt Defteri seçenekleri

Aşağıdaki seçenekler şema kayıt defteri kullanımına özgüdür ve genel seçenekler başvurusunda ele alınmaz.

Option	Zorunlu	Varsayılan	Description
`schema.registry.schema.evolution.mode`	No	`"restart"`	Gelen kayıtta daha yeni bir şema kimliği algılandığında şema değişiklikleri nasıl işlenir? `"restart"`, sorguyu bir `UnknownFieldException` ile sonlandırır; değişikliklerin alınabilmesi için işleri hata durumunda yeniden başlayacak şekilde yapılandırın. `"none"` şema kimliği değişikliklerini yoksayar ve özgün şemayla daha yeni kayıtları ayrıştırıyor.
`confluent.schema.registry.<option>`	No	—	`"confluent.schema.registry"` ön ekini kullanarak herhangi bir Confluent Schema Registry istemcisi seçeneğini iletin. Örneğin, basic auth’u yapılandırmak için `"confluent.schema.registry.basic.auth.credentials.source"` öğesini `"USER_INFO"` olarak ve `"confluent.schema.registry.basic.auth.user.info"` öğesini `"<KEY>:<SECRET>"` olarak ayarlayın.

Usage

Aşağıdaki örneklerde, Apache Spark struct’larını to_protobuf() ile ikili protobuf’a serileştirmeyi ve ikili protobuf kayıtlarını from_protobuf() ile seriden çıkarmayı göstermek için Wanderbricks veri kümesi kullanılır.

Confluent Schema Registry ile protobuf kullanma

Azure Databricks, Protobuf tanımlamak için Confluent Schema Registry'nin kullanılmasını destekler.

Piton

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 programlama dili

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

Dış Confluent Schema Registry'de Kimlik Doğrulaması

Dış Confluent Schema Registry'de kimlik doğrulaması yapmak için şema kayıt defteri seçeneklerinizi kimlik doğrulama kimlik bilgilerini ve API anahtarlarını içerecek şekilde güncelleştirin.

Piton

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 programlama dili

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

Unity Kataloğu birimlerinde truststore ve keystore dosyalarını kullanma

Databricks Runtime 14.3 LTS ve üzerinde, Confluent Schema Registry'de kimlik doğrulaması yapmak için Unity Kataloğu birimlerindeki truststore ve keystore dosyalarını kullanabilirsiniz. Şema kayıt defteri seçeneklerinizi aşağıdaki örneğe göre güncelleştirin:

Piton

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 programlama dili

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

Açıklayıcı dosyayla Protobuf kullanma

İşlem kümenizde kullanılabilen bir protobuf tanımlayıcı dosyasına da başvurabilirsiniz. Konumuna bağlı olarak dosyayı okumak için uygun izinlere sahip olduğunuzdan emin olun.

Piton

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 programlama dili

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

Ek kaynaklar

Akış Avro verilerini okuma ve yazma: Akış iş yükünüz Protobuf yerine Avro serileştirmesi kullanıyorsa eşdeğer from_avro ve to_avro işlevleri için Avro akış işlevlerine bakın.

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?

Last updated on 2026-06-24