Spark结构化流中如何正确映射消息为带schema和payload的对象？

问题分析 #

报错的核心原因是你尝试直接使用Spark的DecimalType对象构造消息中的schema字段，但Spark无法将DecimalType作为字面量序列化到JSON中。Aiven JDBC Sink需要的是字符串形式的类型标识（如"decimal"），而非Spark内部的类型对象。

实现思路 #

不要依赖Spark的StructType来构建schema部分，而是直接用字符串字面量构造嵌套的JSON结构：

1. 构造包含type和fields的schema结构体，其中每个字段的type用字符串"decimal"（可指定精度如"decimal(18,9)"）
2. 直接复用原Parquet数据作为payload部分
3. 将整个schema+payload的结构体转为JSON字符串，作为Kafka消息的value字段

代码示例（Scala） #

import org.apache.spark.sql.functions._

// 读取S3上的Parquet文件
val rawDF = spark.readStream.parquet("s3://your-parquet-path")

// 构造消息中的schema部分
val schemaComponent = struct(
  lit("struct").alias("type"),
  array(
    struct(lit("decimal").alias("type"), lit(false).alias("optional"), lit("timestamp").alias("field")),
    struct(lit("decimal").alias("type"), lit(true).alias("optional"), lit("current").alias("field")),
    struct(lit("decimal").alias("type"), lit(true).alias("optional"), lit("voltage").alias("field")),
    struct(lit("decimal").alias("type"), lit(true).alias("optional"), lit("temperature").alias("field"))
  ).alias("fields")
).alias("schema")

// 构造消息中的payload部分（直接复用原表字段）
val payloadComponent = struct(
  col("timestamp"),
  col("current"),
  col("voltage"),
  col("temperature")
).alias("payload")

// 组合成最终消息并转为JSON字符串
val kafkaOutputDF = rawDF.select(to_json(struct(schemaComponent, payloadComponent)).alias("value"))

// 写入Kafka
kafkaOutputDF.writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "your-kafka-broker:port")
  .option("topic", "your-target-topic")
  .option("checkpointLocation", "s3://your-checkpoint-path")
  .start()
  .awaitTermination()

代码示例（Python） #

from pyspark.sql import functions as F

# 读取Parquet文件
raw_df = spark.readStream.parquet("s3://your-parquet-path")

# 构造schema部分
schema_component = F.struct(
    F.lit("struct").alias("type"),
    F.array(
        F.struct(F.lit("decimal").alias("type"), F.lit(False).alias("optional"), F.lit("timestamp").alias("field")),
        F.struct(F.lit("decimal").alias("type"), F.lit(True).alias("optional"), F.lit("current").alias("field")),
        F.struct(F.lit("decimal").alias("type"), F.lit(True).alias("optional"), F.lit("voltage").alias("field")),
        F.struct(F.lit("decimal").alias("type"), F.lit(True).alias("optional"), F.lit("temperature").alias("field"))
    ).alias("fields")
).alias("schema")

# 构造payload部分
payload_component = F.struct(
    F.col("timestamp"),
    F.col("current"),
    F.col("voltage"),
    F.col("temperature")
).alias("payload")

# 生成最终Kafka消息
kafka_output_df = raw_df.select(F.to_json(F.struct(schema_component, payload_component)).alias("value"))

# 写入Kafka
kafka_output_df.writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your-kafka-broker:port") \
    .option("topic", "your-target-topic") \
    .option("checkpointLocation", "s3://your-checkpoint-path") \
    .start() \
    .awaitTermination()

关键注意事项 #

1. 类型匹配：如果需要指定decimal的精度和小数位，可将schema中的type改为lit("decimal(18,9)")（根据实际业务调整精度）
2. 字段一致性：schema中的field名称必须和payload中的字段完全一致，否则JDBC Sink无法正确映射
3. Connector配置：确保Aiven JDBC Sink Connector的value.converter设置为org.apache.kafka.connect.json.JsonConverter，且value.converter.schemas.enable=true（因为消息本身包含schema结构）

内容的提问来源于stack exchange，提问作者Hongbo Miao