PySpark读取Debezium Kafka Avro数据乱码，寻求解析方案

1. 乱码是否影响目标库加载？ #

会严重影响。你看到的"乱码"是Avro序列化后的二进制字节流，PySpark默认将其转换为字符串后显示为乱码，但本质是未解析的二进制数据。如果直接将这种数据写入目标MariaDB，只会存储无意义的乱码内容，无法得到结构化的业务数据，完全不符合同步需求。

2. PySpark中正常解析Avro数据的实现方案 #

因为Debezium通过Schema Registry将数据序列化为Avro格式，所以PySpark需要借助Avro解析库结合Schema Registry来解析数据，步骤如下：

步骤1：确保PySpark环境包含依赖 #

运行PySpark时需要加载Spark Avro库和Confluent Schema Registry客户端依赖，示例启动命令：

pyspark --packages org.apache.spark:spark-avro_2.12:3.5.0,io.confluent:kafka-avro-serializer:7.4.0

注意：版本需与你的Spark、Schema Registry版本匹配，比如Spark 3.5对应Confluent 7.4系列。

步骤2：修改PySpark代码解析Avro数据 #

以下是完整的流处理解析示例（批量处理可替换readStream/writeStream为read/write）：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_avro, col
from confluent_kafka.schema_registry import SchemaRegistryClient

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("DebeziumAvroParser") \
    .getOrCreate()

# 配置Kafka和Schema Registry地址
kafka_bootstrap_servers = "kafka:9092"
schema_registry_url = "http://schema-registry:8081"
kafka_topic = "mariadb.inventory.customers"

# 读取Kafka原始流
kafka_raw_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", kafka_bootstrap_servers) \
    .option("subscribe", kafka_topic) \
    .load()

# 从Schema Registry动态获取主题对应的最新Avro Schema
schema_client = SchemaRegistryClient({"url": schema_registry_url})
subject_name = f"{kafka_topic}-value"  # Debezium默认主题对应的Schema Subject格式为<topic>-value
latest_schema = schema_client.get_latest_version(subject_name).schema.schema_str

# 解析Avro数据，提取Debezium封装的业务数据
parsed_df = kafka_raw_df.select(
    # 解析value字段的Avro数据
    from_avro(col("value"), latest_schema, schema_registry_url).alias("debezium_msg")
).select(
    # Debezium的Avro结构中，更新后的业务数据存放在payload.after下
    "debezium_msg.payload.after.*"
)

# 将解析后的结构化数据写入目标MariaDB
write_query = parsed_df.writeStream \
    .format("jdbc") \
    .option("url", "jdbc:mariadb://target-mariadb:3306/inventory") \
    .option("dbtable", "customers_sync") \
    .option("user", "root") \
    .option("password", "your_password") \
    .option("checkpointLocation", "/tmp/spark_checkpoint") \
    .start()

write_query.awaitTermination()

关键注意事项 #

• Debezium的Avro结构：默认消息包含schema和payload两层，payload内的after字段对应数据库更新后的最新数据，before为更新前数据，按需提取即可。
• Schema Registry权限：如果Schema Registry开启了认证，需在SchemaRegistryClient配置中添加basic.auth.user.info等参数。
• 依赖兼容性：确保Spark Avro库的Scala版本与你的Spark环境一致（比如_2.12对应Scala 2.12）。

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