Apache Spark Protobuf 数据源完全指南to_protobuf 与 from_protobuf 的使用、类型映射与配置详解【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark自 Spark 3.4.0 起Spark SQL 为 Protobuf 数据提供了内置读写支持由独立的外部模块spark-protobuf提供to_protobuf()与from_protobuf()两个内置函数。本文以当前仓库中的官方文档 docs/sql-data-sources-protobuf.md 为主体结合connector/protobuf模块的源码实现完整讲解该模块的部署方式、两种 schema 指定方式、Protobuf 与 Spark SQL 类型的双向映射、循环引用处理以及全部数据源选项并给出可直接运行的 Kafka Protobuf 流处理示例帮助你掌握在 Spark 批处理与流处理场景中编码、解码 Protobuf 数据的完整实战方案。模块部署spark-protobuf 需要显式引入spark-protobuf是一个外部模块默认不会被spark-submit或spark-shell加载。因此在使用to_protobuf()/from_protobuf()之前必须先通过--packages参数显式引入对应版本的 jar 及其依赖。以当前仓库connector/protobuf/pom.xml中的 artifact 定义为准模块坐标为org.apache.spark:spark-protobuf_2.13${scala.binary.version}占位符在构建时替换为实际的 Scala 二进制版本如2.13版本号与当前 Spark 版本一致。提交应用程序时使用spark-submit./bin/spark-submit --packages org.apache.spark:spark-protobuf_2.13:5.0.0-SNAPSHOT ...在spark-shell中做实验时同样使用--packages./bin/spark-shell --packages org.apache.spark:spark-protobuf_2.13:5.0.0-SNAPSHOT ...提示实际使用时应将版本号替换为你所部署的 Spark 发行版本。更多关于携带外部依赖提交应用的方法参见 Application Submission Guide。自行构建 spark-protobuf 模块当前仓库中spark-protobuf模块源码位于 connector/protobuf其开发者文档 connector/protobuf/README.md 给出了两种构建方式./build/mvn clean package # 或 ./build/sbt clean package当编译环境无法使用官方protoc二进制例如 CentOS 6/7 的glibc版本低于 2.14时可以指定用户自定义的protocexport SPARK_PROTOC_EXEC_PATH/path-to-protoc-exe ./build/mvn -Phive -Puser-defined-protoc clean packageconnector/protobuf/pom.xml中配置了default-protoc默认激活与user-defined-protoc两个 profile后者通过环境变量SPARK_PROTOC_EXEC_PATH注入protoc可执行文件路径用于从src/test/resources/protobuf下的.proto文件生成 Java 类与 descriptor 文件。核心函数to_protobuf() 与 from_protobuf()spark-protobuf包提供两个核心函数from_protobuf()将二进制 Protobuf 数据解码为 Spark SQL 列to_protobuf()将 Spark SQL 列编码为二进制 Protobuf 格式。两个函数都是一列进、一列出的列变换函数输入/输出的 SQL 数据类型既可以是复杂类型Struct、Array、Map也可以是基本类型。Spark SQL 的 schema 是根据传给函数的Protobuf descriptor 文件或Protobuf Java 类自动生成的。重要约束指定的 Protobuf 类或 descriptor 文件必须与数据实际结构一致否则行为未定义——可能解析失败也可能返回任意结果。为什么在 Kafka 流场景中特别有用将 Protobuf 消息作为列处理在读写 Kafka 这类流式数据源时非常实用。Kafka 的每条 key-value 记录会附带元数据如写入 Kafka 的 ingestion 时间戳、offset 等如果承载业务数据的value字段是 Protobuf 格式可以用from_protobuf()将其解出再执行数据富化enrich、清洗clean最后写回 Kafka 或输出到其他 sinkto_protobuf()则用于把多个结构体字段重新编码为单个 Protobuf 消息特别适合在写出到 Kafka 前将多列压缩为一列。方式一通过 Protobuf descriptor 文件指定 schemadescriptor 文件可通过protoc命令根据.proto文件生成。以下使用一个简单的proto3消息syntax proto3; message AppEvent { string name 1; int64 id 2; string context 3; }Python 示例from pyspark.sql.protobuf.functions import from_protobuf, to_protobuf # from_protobuf 和 to_protobuf 提供两种 schema 指定方式 # 一是 Protobuf descriptor 文件二是 shaded Java 类。 df spark \ .readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(subscribe, topic1) \ .load() # 1. 将 schema 为 AppEvent 的 Protobuf 数据解码为 struct # 2. 按列 name 过滤 # 3. 将列 event 编码为 Protobuf 格式。 output df \ .select(from_protobuf(value, AppEvent, descriptorFilePath).alias(event)) \ .where(event.name alice) \ .select(to_protobuf(event, AppEvent, descriptorFilePath).alias(event)) output.printSchema() # root # |--event: struct (nullable true) # | |-- name : string (nullable true) # | |-- id: long (nullable true) # | |-- context: string (nullable true) query output \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(topic, topic2) \ .start()Scala 示例import org.apache.spark.sql.protobuf.functions._ val df spark .readStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(subscribe, topic1) .load() val output df .select(from_protobuf($value, AppEvent, descriptorFilePath) as $event) .where(event.name \alice\) .select(to_protobuf($event, AppEvent, descriptorFilePath) as $event) val query output .writeStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start()Java 示例import static org.apache.spark.sql.functions.col; import static org.apache.spark.sql.protobuf.functions.*; DatasetRow df spark .readStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(subscribe, topic1) .load(); DatasetRow output df .select(from_protobuf(col(value), AppEvent, descriptorFilePath).as(event)) .where(event.name \alice\) .select(to_protobuf(col(event), AppEvent, descriptorFilePath).as(event)); StreamingQuery query output .writeStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start();方式二通过 shaded Java 类指定 schema除 descriptor 文件外也可以直接传入 Protobuf 类名来解码/编码。关键前提指定的 Protobuf 类必须与数据匹配否则行为未定义——可能失败或返回任意结果。同时为避免与用户自身依赖中的com.google.protobuf.*类发生冲突包含这些类的 jar必须做 shaded重定位处理。当前仓库正是这样做的connector/protobuf/pom.xml中配置了maven-shade-plugin将com.google.protobuf重定位为${spark.shade.packageName}.spark_protobuf.protobuf。因此官方文档示例中的类名为org.sparkproject.spark_protobuf.protobuf.AppEvent其中org.sparkproject即spark.shade.packageName的取值。Python 示例from pyspark.sql.protobuf.functions import from_protobuf, to_protobuf output df \ .select(from_protobuf(value, org.sparkproject.spark_protobuf.protobuf.AppEvent).alias(event)) \ .where(event.name alice) output.printSchema() # root # |--event: struct (nullable true) # | |-- name : string (nullable true) # | |-- id: long (nullable true) # | |-- context: string (nullable true) output output \ .select(to_protobuf(event, org.sparkproject.spark_protobuf.protobuf.AppEvent).alias(event)) query output \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(topic, topic2) \ .start()Scala 示例import org.apache.spark.sql.protobuf.functions._ var output df .select(from_protobuf($value, org.example.protos.AppEvent) as $event) .where(event.name \alice\) output.printSchema() // root // |--event: struct (nullable true) // | |-- name : string (nullable true) // | |-- id: long (nullable true) // | |-- context: string (nullable true) output output.select(to_protobuf($event, org.sparkproject.spark_protobuf.protobuf.AppEvent) as $event) val query output .writeStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start()Java 示例DatasetRow output df .select( from_protobuf(col(value), org.sparkproject.spark_protobuf.protobuf.AppEvent).as(event)) .where(event.name \alice\); output.printSchema(); // root // |--event: struct (nullable true) // | |-- name : string (nullable true) // | |-- id: long (nullable true) // | |-- context: string (nullable true) output output.select( to_protobuf(col(event), org.sparkproject.spark_protobuf.protobuf.AppEvent).as(event)); StreamingQuery query output .writeStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start();底层实现原理源码视角从源码看这两个函数分别是两个 Catalyst 表达式的高层封装from_protobuf()→ProtobufDataToCatalystconnector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDataToCatalyst.scala输入类型固定为BinaryType输出类型由SchemaConverters.toSqlType根据消息 descriptor 在编译期推导因此from_protobuf()之后的列类型是静态可知的运行时通过DynamicMessage.parseFrom(messageDescriptor, binary, extensionRegistry)解析二进制数据再交给ProtobufDeserializer转成 Catalyst 内部行解析时还会检查未知字段中是否包含与已知常规字段同号的字段——这通常意味着读写双方 schema 不匹配此时会抛出protobufFieldTypeMismatchErrorparseMode在编译期校验仅接受PERMISSIVE与FAILFASTDROPMALFORMED会被拒绝见下文mode选项说明。to_protobuf()→CatalystDataToProtobufconnector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/CatalystDataToProtobuf.scala输出类型固定为BinaryType运行时通过ProtobufSerializer将 Catalyst 行序列化为DynamicMessage再调用toByteArray得到最终字节流源码注释中留有 TODOSPARK-43578descriptor 文件可能很大未来应考虑广播以避免随每个 task 传输。descriptor 的构建与字段匹配逻辑集中在 connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala 中的buildDescriptor与ProtoSchemaHelper后者会同时从 Protobuf descriptor 的常规字段与 extension 字段中按名称大小写不敏感与 Catalyst schema 做字段匹配并在序列化时校验Catalyst 中存在但 Protobuf 中不存在的多余字段。Protobuf → Spark SQL 类型转换from_protobuf()解码时支持 Protobuf 的[标量类型]、[enum]、[嵌套类型]与[map]类型除此之外spark-protobuf还引入了对 ProtobufOneOf字段的支持——OneOf允许一个消息拥有多组字段、但任意时刻只能有一组存在。这在数据格式不固定、需要无报错地处理多种字段组合的场景下非常有用。完整映射关系如下Protobuf 类型Spark SQL 类型booleanBooleanTypeintIntegerTypelongLongTypefloatFloatTypedoubleDoubleTypestringStringTypeenumStringTypebytesBinaryTypeMessageStructTyperepeatedArrayTypemapMapTypeOneOfStruct对 google.protobuf 逻辑类型的支持spark-protobuf还支持读取 Protobuf 的google.protobuf.Timestamp与google.protobuf.Duration两种 well-known 逻辑类型Protobuf 逻辑类型Protobuf schemaSpark SQL 类型durationMessageType{seconds: Long, nanos: Int}DayTimeIntervalTypetimestampMessageType{seconds: Long, nanos: Int}TimestampTypeSpark SQL → Protobuf 类型转换to_protobuf()支持将所有Spark SQL 类型写入 Protobuf。对大多数类型而言映射是直观的例如IntegerType转为intSpark SQL 类型Protobuf 类型BooleanTypebooleanIntegerTypeintLongTypelongFloatTypefloatDoubleTypedoubleStringTypestringStringTypeenumBinaryTypebytesStructTypemessageArrayTyperepeatedMapTypemapTimeTypeint64 (nanoseconds-of-day)TimeType的特别说明TimeType会被写为普通的 Protobufint64其值为自午夜以来的纳秒数。由于裸的int64不携带逻辑类型标记from_protobuf()会将该字段读回为LongType即纳秒数如需还原为 TIME 值可用cast(... AS TIME)进行转换。处理 Protobuf 循环引用字段使用 Protobuf 时一个常见问题是循环引用某个字段引用自身或引用另一个最终又指回原字段的字段。如果不加处理解析 schema 时可能导致无限递归或其他异常。spark-protobuf通过recursive.fields.max.depth选项支持按字段类型检查循环引用用于指定解析 schema 时允许的最大递归层数默认值为-1表示不允许任何递归字段可设置为1 到 10设为1丢弃所有递归字段设为2允许递归一层设为3允许递归两层依此类推不允许大于 10否则可能引发性能问题甚至栈溢出。例如考虑如下proto3消息syntax proto3; message Person { string name 1; Person bff 2; }根据recursive.fields.max.depth的取值上述 schema 会转换为不同的 Spark SQL 列结构1: structname: string 2: structname: string, bff: structname: string 3: structname: string, bff: structname: string, bff: structname: string ...源码中 ProtobufOptions.scala 对该选项的注释进一步说明当消息深度超过该上限时返回的 Spark struct 会在递归上限处被截断从而避免生成过大的 schema。数据源选项Data Source OptionProtobuf 相关的数据源选项通过内置函数from_protobuf/to_protobuf设置。选项名大小写不敏感源码中ProtobufOptions基于CaseInsensitiveMap解析。完整选项如下属性名默认值含义作用域modeFAILFAST解析损坏记录时的处理模式。PERMISSIVE遇到损坏记录时将所有字段置为nullDROPMALFORMED忽略整条损坏记录该模式在 Protobuf 内置函数中不受支持FAILFAST遇到损坏记录时抛出异常。readrecursive.fields.max.depth-1解析 schema 时允许的最大递归层数用于处理循环引用字段见上文合法范围为-1与1~10。readconvert.any.fields.to.jsonfalse是否将 ProtobufAny字段转换为 JSON。该选项应谨慎启用JSON 转换与处理效率较低且会降低 schema 安全性使下游处理更容易出错。reademit.default.valuesfalse反序列化 Protobuf 到 Spark struct 时是否渲染零值字段。序列化的 Protobuf 中某字段为空时默认反序列化为null开启后渲染类型对应的零值。readenums.as.intsfalse是否将 enum 字段渲染为其整数值。为false时 enum 映射为StringType值为枚举名为true时映射为IntegerType值为枚举整数值。readupcast.unsigned.intsfalse是否将无符号整数升级为更大的类型。为true时uint32使用LongType、uint64使用Decimal(20, 0)从而在不溢出的前提下容纳大无符号值。readunwrap.primitive.wrapper.typesfalse反序列化时是否解包 well-known 基本类型包装类如google.protobuf.Int32Value、google.protobuf.Int64Value等的 struct 表示。默认这些包装类会被反序列化为 struct。readretain.empty.message.typesfalse是否在 Schema 中保留空 proto 消息类型的字段。由于 Spark 不允许写出空StructType空 proto 消息默认会被丢弃设为true时会向空 proto 消息插入一个哑列__dummy_field_in_empty_struct从而保留空消息字段。read关键选项的源码级细节结合 connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala 的注释可以进一步理解几个易混淆选项的行为modefrom_protobuf对应的ProtobufDataToCatalyst在编译期校验解析模式仅接受PERMISSIVE与FAILFAST若传入DROPMALFORMED会直接抛出parseModeUnsupportedError。因此表中明确标注DROPMALFORMED不受支持。convert.any.fields.to.json未开启时Any字段表现为一个含两个字段的普通消息——STRUCTtype_url: STRING, value: BINARY其中二进制value不做解释实践中不便使用。开启后 schema 中该字段变为STRING运行时将Any内的真实消息解析并转为 JSON 字符串形如{type:type.googleapis.com/...,name:Mario,id:100}。使用时需注意两点一是 descriptor 文件中必须包含Any字段可能出现的所有 Protobuf 类型定义否则该记录会报错二是该特性也支持 Java 类方式但只能看到与主 Java 类同一个.proto文件中定义的类型。emit.default.values该选项行为类似protobuf-java-util的JsonFormat.includingDefaultValues或golang/protobuf的jsonpb.emitDefaults。例如 proto3 消息Person { string name1; int64 age2; optional string middle_name3; optional int64 salary4; }当构造Person(age0, middle_name)时不开此选项from_protobuf()结果中age为null因为 proto3 中零值单数域不出现在线上格式里开启后结果为name、age0而显式声明为optional的salary仍为null。enums.as.ints例如Person { enum Job { NONE0; ENGINEER1; DOCTOR2; } Job job1; }对Person(jobENGINEER)默认反序列化为{job: ENGINEER}开启后为{job: 1}且输出列的 SQL 类型会从string变为int需要注意对既有解析逻辑的影响。upcast.unsigned.ints默认情况下uint32/uint64分别序列化为有符号的IntegerType/LongType当无符号值过大uint32超过 2^31、uint64超过 2^63时会溢出为负数开启该选项后通过扩大类型避免溢出。unwrap.primitive.wrapper.types例如message Example { google.protobuf.Int32Value int_val 1; }默认Example(Int32Value(5))被反序列化为{int_val: {value: 5}}开启后为{int_val: 5}。注意该选项与emit.default.values同时开启时不会在解包时填充默认基本值以尽量保留信息。retain.empty.message.types例如message A {}与message B { A a 1; string name 2; }默认情况下字段a被丢弃schema 变为structname: string设为true后通过插入哑列保留schema 变为structa: struct__dummy_field_in_empty_struct: string, name: string。测试与验证当前仓库connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/下提供了完整的测试套件可作为理解各选项实际行为与验证用法的参考ProtobufFunctionsSuite.scalafrom_protobuf/to_protobuf函数的基本功能测试ProtobufSerdeSuite.scala序列化/反序列化Serde测试ProtobufDescriptorFileReadSuite.scala基于 descriptor 文件读取的测试ProtobufCatalystDataConversionSuite.scalaCatalyst 与 Protobuf 数据双向转换测试ProtobufExtensionsSuite.scalaProtobuf 扩展extension字段相关测试ProtobufTestBase.scala测试基类与公共工具。小结spark-protobuf模块让 Spark SQL 从 3.4.0 起原生支持 Protobuf 数据的读写。本文覆盖了从模块部署--packages引入spark-protobuf_2.13、两种 schema 指定方式descriptor 文件与 shaded Java 类、三语言 API 用法、双向类型映射、OneOf与 well-known 逻辑类型支持、循环引用处理recursive.fields.max.depth到全部 8 个数据源选项的完整语义。结合 connector/protobuf 模块的源码ProtobufOptions.scala、ProtobufDataToCatalyst.scala、CatalystDataToProtobuf.scala、ProtobufUtils.scala你可以进一步深入理解 schema 推导、DynamicMessage解析、shaded 类机制与字段匹配校验等底层原理进而在生产环境尤其是 Kafka 流处理中可靠地落地 Protobuf 数据管道。【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Apache Spark Protobuf 数据源完全指南：to_protobuf 与 from_protobuf 的使用、类型映射与配置详解
Apache Spark Protobuf 数据源完全指南to_protobuf 与 from_protobuf 的使用、类型映射与配置详解【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark自 Spark 3.4.0 起Spark SQL 为 Protobuf 数据提供了内置读写支持由独立的外部模块spark-protobuf提供to_protobuf()与from_protobuf()两个内置函数。本文以当前仓库中的官方文档 docs/sql-data-sources-protobuf.md 为主体结合connector/protobuf模块的源码实现完整讲解该模块的部署方式、两种 schema 指定方式、Protobuf 与 Spark SQL 类型的双向映射、循环引用处理以及全部数据源选项并给出可直接运行的 Kafka Protobuf 流处理示例帮助你掌握在 Spark 批处理与流处理场景中编码、解码 Protobuf 数据的完整实战方案。模块部署spark-protobuf 需要显式引入spark-protobuf是一个外部模块默认不会被spark-submit或spark-shell加载。因此在使用to_protobuf()/from_protobuf()之前必须先通过--packages参数显式引入对应版本的 jar 及其依赖。以当前仓库connector/protobuf/pom.xml中的 artifact 定义为准模块坐标为org.apache.spark:spark-protobuf_2.13${scala.binary.version}占位符在构建时替换为实际的 Scala 二进制版本如2.13版本号与当前 Spark 版本一致。提交应用程序时使用spark-submit./bin/spark-submit --packages org.apache.spark:spark-protobuf_2.13:5.0.0-SNAPSHOT ...在spark-shell中做实验时同样使用--packages./bin/spark-shell --packages org.apache.spark:spark-protobuf_2.13:5.0.0-SNAPSHOT ...提示实际使用时应将版本号替换为你所部署的 Spark 发行版本。更多关于携带外部依赖提交应用的方法参见 Application Submission Guide。自行构建 spark-protobuf 模块当前仓库中spark-protobuf模块源码位于 connector/protobuf其开发者文档 connector/protobuf/README.md 给出了两种构建方式./build/mvn clean package # 或 ./build/sbt clean package当编译环境无法使用官方protoc二进制例如 CentOS 6/7 的glibc版本低于 2.14时可以指定用户自定义的protocexport SPARK_PROTOC_EXEC_PATH/path-to-protoc-exe ./build/mvn -Phive -Puser-defined-protoc clean packageconnector/protobuf/pom.xml中配置了default-protoc默认激活与user-defined-protoc两个 profile后者通过环境变量SPARK_PROTOC_EXEC_PATH注入protoc可执行文件路径用于从src/test/resources/protobuf下的.proto文件生成 Java 类与 descriptor 文件。核心函数to_protobuf() 与 from_protobuf()spark-protobuf包提供两个核心函数from_protobuf()将二进制 Protobuf 数据解码为 Spark SQL 列to_protobuf()将 Spark SQL 列编码为二进制 Protobuf 格式。两个函数都是一列进、一列出的列变换函数输入/输出的 SQL 数据类型既可以是复杂类型Struct、Array、Map也可以是基本类型。Spark SQL 的 schema 是根据传给函数的Protobuf descriptor 文件或Protobuf Java 类自动生成的。重要约束指定的 Protobuf 类或 descriptor 文件必须与数据实际结构一致否则行为未定义——可能解析失败也可能返回任意结果。为什么在 Kafka 流场景中特别有用将 Protobuf 消息作为列处理在读写 Kafka 这类流式数据源时非常实用。Kafka 的每条 key-value 记录会附带元数据如写入 Kafka 的 ingestion 时间戳、offset 等如果承载业务数据的value字段是 Protobuf 格式可以用from_protobuf()将其解出再执行数据富化enrich、清洗clean最后写回 Kafka 或输出到其他 sinkto_protobuf()则用于把多个结构体字段重新编码为单个 Protobuf 消息特别适合在写出到 Kafka 前将多列压缩为一列。方式一通过 Protobuf descriptor 文件指定 schemadescriptor 文件可通过protoc命令根据.proto文件生成。以下使用一个简单的proto3消息syntax proto3; message AppEvent { string name 1; int64 id 2; string context 3; }Python 示例from pyspark.sql.protobuf.functions import from_protobuf, to_protobuf # from_protobuf 和 to_protobuf 提供两种 schema 指定方式 # 一是 Protobuf descriptor 文件二是 shaded Java 类。 df spark \ .readStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(subscribe, topic1) \ .load() # 1. 将 schema 为 AppEvent 的 Protobuf 数据解码为 struct # 2. 按列 name 过滤 # 3. 将列 event 编码为 Protobuf 格式。 output df \ .select(from_protobuf(value, AppEvent, descriptorFilePath).alias(event)) \ .where(event.name alice) \ .select(to_protobuf(event, AppEvent, descriptorFilePath).alias(event)) output.printSchema() # root # |--event: struct (nullable true) # | |-- name : string (nullable true) # | |-- id: long (nullable true) # | |-- context: string (nullable true) query output \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(topic, topic2) \ .start()Scala 示例import org.apache.spark.sql.protobuf.functions._ val df spark .readStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(subscribe, topic1) .load() val output df .select(from_protobuf($value, AppEvent, descriptorFilePath) as $event) .where(event.name \alice\) .select(to_protobuf($event, AppEvent, descriptorFilePath) as $event) val query output .writeStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start()Java 示例import static org.apache.spark.sql.functions.col; import static org.apache.spark.sql.protobuf.functions.*; DatasetRow df spark .readStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(subscribe, topic1) .load(); DatasetRow output df .select(from_protobuf(col(value), AppEvent, descriptorFilePath).as(event)) .where(event.name \alice\) .select(to_protobuf(col(event), AppEvent, descriptorFilePath).as(event)); StreamingQuery query output .writeStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start();方式二通过 shaded Java 类指定 schema除 descriptor 文件外也可以直接传入 Protobuf 类名来解码/编码。关键前提指定的 Protobuf 类必须与数据匹配否则行为未定义——可能失败或返回任意结果。同时为避免与用户自身依赖中的com.google.protobuf.*类发生冲突包含这些类的 jar必须做 shaded重定位处理。当前仓库正是这样做的connector/protobuf/pom.xml中配置了maven-shade-plugin将com.google.protobuf重定位为${spark.shade.packageName}.spark_protobuf.protobuf。因此官方文档示例中的类名为org.sparkproject.spark_protobuf.protobuf.AppEvent其中org.sparkproject即spark.shade.packageName的取值。Python 示例from pyspark.sql.protobuf.functions import from_protobuf, to_protobuf output df \ .select(from_protobuf(value, org.sparkproject.spark_protobuf.protobuf.AppEvent).alias(event)) \ .where(event.name alice) output.printSchema() # root # |--event: struct (nullable true) # | |-- name : string (nullable true) # | |-- id: long (nullable true) # | |-- context: string (nullable true) output output \ .select(to_protobuf(event, org.sparkproject.spark_protobuf.protobuf.AppEvent).alias(event)) query output \ .writeStream \ .format(kafka) \ .option(kafka.bootstrap.servers, host1:port1,host2:port2) \ .option(topic, topic2) \ .start()Scala 示例import org.apache.spark.sql.protobuf.functions._ var output df .select(from_protobuf($value, org.example.protos.AppEvent) as $event) .where(event.name \alice\) output.printSchema() // root // |--event: struct (nullable true) // | |-- name : string (nullable true) // | |-- id: long (nullable true) // | |-- context: string (nullable true) output output.select(to_protobuf($event, org.sparkproject.spark_protobuf.protobuf.AppEvent) as $event) val query output .writeStream .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start()Java 示例DatasetRow output df .select( from_protobuf(col(value), org.sparkproject.spark_protobuf.protobuf.AppEvent).as(event)) .where(event.name \alice\); output.printSchema(); // root // |--event: struct (nullable true) // | |-- name : string (nullable true) // | |-- id: long (nullable true) // | |-- context: string (nullable true) output output.select( to_protobuf(col(event), org.sparkproject.spark_protobuf.protobuf.AppEvent).as(event)); StreamingQuery query output .writeStream() .format(kafka) .option(kafka.bootstrap.servers, host1:port1,host2:port2) .option(topic, topic2) .start();底层实现原理源码视角从源码看这两个函数分别是两个 Catalyst 表达式的高层封装from_protobuf()→ProtobufDataToCatalystconnector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDataToCatalyst.scala输入类型固定为BinaryType输出类型由SchemaConverters.toSqlType根据消息 descriptor 在编译期推导因此from_protobuf()之后的列类型是静态可知的运行时通过DynamicMessage.parseFrom(messageDescriptor, binary, extensionRegistry)解析二进制数据再交给ProtobufDeserializer转成 Catalyst 内部行解析时还会检查未知字段中是否包含与已知常规字段同号的字段——这通常意味着读写双方 schema 不匹配此时会抛出protobufFieldTypeMismatchErrorparseMode在编译期校验仅接受PERMISSIVE与FAILFASTDROPMALFORMED会被拒绝见下文mode选项说明。to_protobuf()→CatalystDataToProtobufconnector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/CatalystDataToProtobuf.scala输出类型固定为BinaryType运行时通过ProtobufSerializer将 Catalyst 行序列化为DynamicMessage再调用toByteArray得到最终字节流源码注释中留有 TODOSPARK-43578descriptor 文件可能很大未来应考虑广播以避免随每个 task 传输。descriptor 的构建与字段匹配逻辑集中在 connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala 中的buildDescriptor与ProtoSchemaHelper后者会同时从 Protobuf descriptor 的常规字段与 extension 字段中按名称大小写不敏感与 Catalyst schema 做字段匹配并在序列化时校验Catalyst 中存在但 Protobuf 中不存在的多余字段。Protobuf → Spark SQL 类型转换from_protobuf()解码时支持 Protobuf 的[标量类型]、[enum]、[嵌套类型]与[map]类型除此之外spark-protobuf还引入了对 ProtobufOneOf字段的支持——OneOf允许一个消息拥有多组字段、但任意时刻只能有一组存在。这在数据格式不固定、需要无报错地处理多种字段组合的场景下非常有用。完整映射关系如下Protobuf 类型Spark SQL 类型booleanBooleanTypeintIntegerTypelongLongTypefloatFloatTypedoubleDoubleTypestringStringTypeenumStringTypebytesBinaryTypeMessageStructTyperepeatedArrayTypemapMapTypeOneOfStruct对 google.protobuf 逻辑类型的支持spark-protobuf还支持读取 Protobuf 的google.protobuf.Timestamp与google.protobuf.Duration两种 well-known 逻辑类型Protobuf 逻辑类型Protobuf schemaSpark SQL 类型durationMessageType{seconds: Long, nanos: Int}DayTimeIntervalTypetimestampMessageType{seconds: Long, nanos: Int}TimestampTypeSpark SQL → Protobuf 类型转换to_protobuf()支持将所有Spark SQL 类型写入 Protobuf。对大多数类型而言映射是直观的例如IntegerType转为intSpark SQL 类型Protobuf 类型BooleanTypebooleanIntegerTypeintLongTypelongFloatTypefloatDoubleTypedoubleStringTypestringStringTypeenumBinaryTypebytesStructTypemessageArrayTyperepeatedMapTypemapTimeTypeint64 (nanoseconds-of-day)TimeType的特别说明TimeType会被写为普通的 Protobufint64其值为自午夜以来的纳秒数。由于裸的int64不携带逻辑类型标记from_protobuf()会将该字段读回为LongType即纳秒数如需还原为 TIME 值可用cast(... AS TIME)进行转换。处理 Protobuf 循环引用字段使用 Protobuf 时一个常见问题是循环引用某个字段引用自身或引用另一个最终又指回原字段的字段。如果不加处理解析 schema 时可能导致无限递归或其他异常。spark-protobuf通过recursive.fields.max.depth选项支持按字段类型检查循环引用用于指定解析 schema 时允许的最大递归层数默认值为-1表示不允许任何递归字段可设置为1 到 10设为1丢弃所有递归字段设为2允许递归一层设为3允许递归两层依此类推不允许大于 10否则可能引发性能问题甚至栈溢出。例如考虑如下proto3消息syntax proto3; message Person { string name 1; Person bff 2; }根据recursive.fields.max.depth的取值上述 schema 会转换为不同的 Spark SQL 列结构1: structname: string 2: structname: string, bff: structname: string 3: structname: string, bff: structname: string, bff: structname: string ...源码中 ProtobufOptions.scala 对该选项的注释进一步说明当消息深度超过该上限时返回的 Spark struct 会在递归上限处被截断从而避免生成过大的 schema。数据源选项Data Source OptionProtobuf 相关的数据源选项通过内置函数from_protobuf/to_protobuf设置。选项名大小写不敏感源码中ProtobufOptions基于CaseInsensitiveMap解析。完整选项如下属性名默认值含义作用域modeFAILFAST解析损坏记录时的处理模式。PERMISSIVE遇到损坏记录时将所有字段置为nullDROPMALFORMED忽略整条损坏记录该模式在 Protobuf 内置函数中不受支持FAILFAST遇到损坏记录时抛出异常。readrecursive.fields.max.depth-1解析 schema 时允许的最大递归层数用于处理循环引用字段见上文合法范围为-1与1~10。readconvert.any.fields.to.jsonfalse是否将 ProtobufAny字段转换为 JSON。该选项应谨慎启用JSON 转换与处理效率较低且会降低 schema 安全性使下游处理更容易出错。reademit.default.valuesfalse反序列化 Protobuf 到 Spark struct 时是否渲染零值字段。序列化的 Protobuf 中某字段为空时默认反序列化为null开启后渲染类型对应的零值。readenums.as.intsfalse是否将 enum 字段渲染为其整数值。为false时 enum 映射为StringType值为枚举名为true时映射为IntegerType值为枚举整数值。readupcast.unsigned.intsfalse是否将无符号整数升级为更大的类型。为true时uint32使用LongType、uint64使用Decimal(20, 0)从而在不溢出的前提下容纳大无符号值。readunwrap.primitive.wrapper.typesfalse反序列化时是否解包 well-known 基本类型包装类如google.protobuf.Int32Value、google.protobuf.Int64Value等的 struct 表示。默认这些包装类会被反序列化为 struct。readretain.empty.message.typesfalse是否在 Schema 中保留空 proto 消息类型的字段。由于 Spark 不允许写出空StructType空 proto 消息默认会被丢弃设为true时会向空 proto 消息插入一个哑列__dummy_field_in_empty_struct从而保留空消息字段。read关键选项的源码级细节结合 connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala 的注释可以进一步理解几个易混淆选项的行为modefrom_protobuf对应的ProtobufDataToCatalyst在编译期校验解析模式仅接受PERMISSIVE与FAILFAST若传入DROPMALFORMED会直接抛出parseModeUnsupportedError。因此表中明确标注DROPMALFORMED不受支持。convert.any.fields.to.json未开启时Any字段表现为一个含两个字段的普通消息——STRUCTtype_url: STRING, value: BINARY其中二进制value不做解释实践中不便使用。开启后 schema 中该字段变为STRING运行时将Any内的真实消息解析并转为 JSON 字符串形如{type:type.googleapis.com/...,name:Mario,id:100}。使用时需注意两点一是 descriptor 文件中必须包含Any字段可能出现的所有 Protobuf 类型定义否则该记录会报错二是该特性也支持 Java 类方式但只能看到与主 Java 类同一个.proto文件中定义的类型。emit.default.values该选项行为类似protobuf-java-util的JsonFormat.includingDefaultValues或golang/protobuf的jsonpb.emitDefaults。例如 proto3 消息Person { string name1; int64 age2; optional string middle_name3; optional int64 salary4; }当构造Person(age0, middle_name)时不开此选项from_protobuf()结果中age为null因为 proto3 中零值单数域不出现在线上格式里开启后结果为name、age0而显式声明为optional的salary仍为null。enums.as.ints例如Person { enum Job { NONE0; ENGINEER1; DOCTOR2; } Job job1; }对Person(jobENGINEER)默认反序列化为{job: ENGINEER}开启后为{job: 1}且输出列的 SQL 类型会从string变为int需要注意对既有解析逻辑的影响。upcast.unsigned.ints默认情况下uint32/uint64分别序列化为有符号的IntegerType/LongType当无符号值过大uint32超过 2^31、uint64超过 2^63时会溢出为负数开启该选项后通过扩大类型避免溢出。unwrap.primitive.wrapper.types例如message Example { google.protobuf.Int32Value int_val 1; }默认Example(Int32Value(5))被反序列化为{int_val: {value: 5}}开启后为{int_val: 5}。注意该选项与emit.default.values同时开启时不会在解包时填充默认基本值以尽量保留信息。retain.empty.message.types例如message A {}与message B { A a 1; string name 2; }默认情况下字段a被丢弃schema 变为structname: string设为true后通过插入哑列保留schema 变为structa: struct__dummy_field_in_empty_struct: string, name: string。测试与验证当前仓库connector/protobuf/src/test/scala/org/apache/spark/sql/protobuf/下提供了完整的测试套件可作为理解各选项实际行为与验证用法的参考ProtobufFunctionsSuite.scalafrom_protobuf/to_protobuf函数的基本功能测试ProtobufSerdeSuite.scala序列化/反序列化Serde测试ProtobufDescriptorFileReadSuite.scala基于 descriptor 文件读取的测试ProtobufCatalystDataConversionSuite.scalaCatalyst 与 Protobuf 数据双向转换测试ProtobufExtensionsSuite.scalaProtobuf 扩展extension字段相关测试ProtobufTestBase.scala测试基类与公共工具。小结spark-protobuf模块让 Spark SQL 从 3.4.0 起原生支持 Protobuf 数据的读写。本文覆盖了从模块部署--packages引入spark-protobuf_2.13、两种 schema 指定方式descriptor 文件与 shaded Java 类、三语言 API 用法、双向类型映射、OneOf与 well-known 逻辑类型支持、循环引用处理recursive.fields.max.depth到全部 8 个数据源选项的完整语义。结合 connector/protobuf 模块的源码ProtobufOptions.scala、ProtobufDataToCatalyst.scala、CatalystDataToProtobuf.scala、ProtobufUtils.scala你可以进一步深入理解 schema 推导、DynamicMessage解析、shaded 类机制与字段匹配校验等底层原理进而在生产环境尤其是 Kafka 流处理中可靠地落地 Protobuf 数据管道。【免费下载链接】sparkApache Spark - A unified analytics engine for large-scale data processing项目地址: https://gitcode.com/gh_mirrors/sp/spark创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考