Baca dan tulis protokol buffer

Buffer Protokol (protobuf) adalah format serialisasi biner netral bahasa yang dikembangkan oleh Google. Azure Databricks pengguna paling sering mengalaminya saat memproses rekaman yang dikodekan biner dari sistem streaming peristiwa seperti Apache Kafka. Azure Databricks mendukung pembacaan serta penulisan data protobuf dengan Apache Spark melalui fungsi from_protobuf dan to_protobuf, yang mengonversi data antara protobuf biner dan tipe struct Spark SQL, baik untuk beban kerja streaming maupun batch.

Prasyarat

Fungsi Protobuf memerlukan Databricks Runtime 12.2 LTS ke atas.

Sintaksis fungsi

Gunakan from_protobuf untuk mentransmisikan kolom biner ke struct, dan to_protobuf untuk mentransmisikan kolom struct ke biner. Anda harus menyediakan file deskriptor yang diidentifikasi oleh descFilePath argumen atau registri skema yang ditentukan dengan options argumen . Untuk daftar lengkap opsi, lihat Protobuf.

Phyton

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

Opsi

Teruskan opsi ke from_protobuf dan to_protobuf menggunakan argumen options. Untuk daftar lengkap opsi yang didukung, lihat Protobuf.

Opsi Registri Skema

Opsi berikut khusus untuk penggunaan registri skema dan tidak tercakup dalam referensi opsi umum.

Option Wajib Default Deskripsi
schema.registry.schema.evolution.mode No "restart" Bagaimana perubahan skema ditangani saat ID skema yang lebih baru terdeteksi dalam rekaman yang masuk. "restart" mengakhiri kueri dengan UnknownFieldException; konfigurasikan job agar dimulai ulang jika gagal untuk menerapkan perubahan. "none" mengabaikan perubahan skema-id dan mengurai rekaman yang lebih baru dengan skema asli.
confluent.schema.registry.<option> No — Berikan opsi klien Confluent Schema Registry menggunakan awalan "confluent.schema.registry". Misalnya, atur "confluent.schema.registry.basic.auth.credentials.source" ke "USER_INFO" dan "confluent.schema.registry.basic.auth.user.info" ke "<KEY>:<SECRET>" untuk mengonfigurasi autentikasi dasar.

Usage

Contoh berikut menggunakan himpunan data Wanderbricks untuk menunjukkan serialisasi struktur Apache Spark ke protobuf biner dengan to_protobuf() dan mendeserialisasi rekaman protobuf biner dengan from_protobuf().

Menggunakan protobuf dengan Confluent Schema Registry

Azure Databricks mendukung penggunaan Confluent Schema Registry untuk mendefinisikan Protobuf.

Phyton

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

Mengautentikasi ke Registri Skema Confluent eksternal

Untuk mengautentikasi ke Registri Skema Confluent eksternal, perbarui opsi registri skema Anda untuk menyertakan kredensial autentikasi dan kunci API.

Phyton

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

Menggunakan file truststore dan keystore pada volume Unity Catalog

Di Databricks Runtime 14.3 LTS ke atas, Anda dapat menggunakan file truststore dan keystore dalam volume Unity Catalog untuk mengautentikasi ke Confluent Schema Registry. Perbarui opsi registri skema Anda sesuai dengan contoh berikut:

Phyton

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

Menggunakan Protobuf dengan file deskriptor

Anda juga dapat mereferensikan file deskriptor protobuf yang tersedia untuk kluster komputasi Anda. Pastikan Anda memiliki izin yang tepat untuk membaca file, bergantung pada lokasinya.

Phyton

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

Sumber daya tambahan

  • Membaca dan menulis data Avro streaming: Jika beban kerja streaming Anda menggunakan serialisasi Avro, bukan Protobuf, lihat fungsi streaming Avro untuk fungsi from_avro dan to_avro yang setara.
  • Last updated on