咨询Pyspark反序列化Kafka中Schema Registry托管的Avro CDC消息方案

一、更简便的替代工具 #

1. Spark官方Confluent Avro集成包 #

Spark原生支持对接Confluent Schema Registry处理Avro数据，无需额外复杂工具，只需引入对应依赖即可实现反序列化：

• 提交Pyspark任务时添加依赖（版本需与你的Spark、Kafka版本匹配）：

pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,io.confluent:kafka-avro-serializer:7.4.0,io.confluent:kafka-schema-registry-client:7.4.0

• 反序列化代码示例：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_avro

spark = SparkSession.builder \
    .appName("KafkaAvroCDCReader") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,io.confluent:kafka-avro-serializer:7.4.0,io.confluent:kafka-schema-registry-client:7.4.0") \
    .getOrCreate()

# 配置Schema Registry地址
schema_registry_config = {"schema.registry.url": "http://your-schema-registry:8081"}

# 读取Kafka流数据
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker:9092") \
    .option("subscribe", "mysql-cdc-topic") \
    .load()

# 反序列化Avro格式的value字段
cdc_df = kafka_df.select(
    col("key").cast("string"),
    from_avro(col("value"), schema_registry_config).alias("cdc_data")
)

# 展开CDC数据结构
final_df = cdc_df.select("key", "cdc_data.*")

# 控制台输出测试
query = final_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

2. fastavro自定义UDF反序列化 #

如果已知Avro Schema，可以用fastavro库编写自定义UDF实现反序列化，适合无需Schema Registry的场景：

• 先安装依赖：pip install fastavro
• 代码示例：

import fastavro
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf, col
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# 定义CDC数据的Avro Schema（可从本地文件或Schema Registry获取）
avro_schema = {
    "type": "record",
    "name": "CDCEvent",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "operation", "type": "string"}
    ]
}

# 编写反序列化UDF
def deserialize_avro(value):
    if value is None:
        return None
    return fastavro.schemaless_reader(value, avro_schema)

deserialize_udf = udf(deserialize_avro, StructType([
    StructField("id", IntegerType()),
    StructField("name", StringType()),
    StructField("operation", StringType())
]))

spark = SparkSession.builder.appName("FastAvroCDCReader").getOrCreate()

# 读取Kafka流
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker:9092") \
    .option("subscribe", "mysql-cdc-topic") \
    .load()

# 应用UDF反序列化
cdc_df = kafka_df.select(
    col("key").cast("string"),
    deserialize_udf(col("value")).alias("cdc_data")
).select("key", "cdc_data.*")

# 控制台输出测试
query = cdc_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

二、ABRiS在Pyspark中的集成方法 #

ABRiS是专门优化Spark Avro数据处理的库，对Confluent Schema Registry支持更完善，集成步骤如下：

1. 引入ABRiS依赖 #

提交Pyspark任务时添加对应maven依赖（版本需与Spark版本匹配，如Spark 3.5对应ABRiS 7.4.0）：

pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,za.co.absa:abris_2.12:7.4.0,io.confluent:kafka-schema-registry-client:7.4.0

或在SparkSession中配置：

spark = SparkSession.builder \
    .appName("ABRiSCDCReader") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,za.co.absa:abris_2.12:7.4.0,io.confluent:kafka-schema-registry-client:7.4.0") \
    .getOrCreate()

2. 反序列化代码实现 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from absa.abris.avro.functions import from_avro

# 配置Schema Registry参数
schema_registry_settings = {
    "schema.registry.url": "http://your-schema-registry:8081",
    "schema.registry.subject": "mysql-cdc-topic-value"  # 对应Kafka Topic的Schema Subject
}

# 读取Kafka流
kafka_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "kafka-broker:9092") \
    .option("subscribe", "mysql-cdc-topic") \
    .load()

# 使用ABRiS的from_avro函数反序列化
cdc_df = kafka_df.select(
    col("key").cast("string"),
    from_avro(col("value"), schema_registry_settings).alias("cdc_data")
)

# 展开CDC数据结构
final_df = cdc_df.select("key", "cdc_data.*")

# 控制台输出测试
query = final_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

注意事项 #

• 务必保证ABRiS、Spark、Confluent相关库的版本兼容，避免出现依赖冲突；
• 若使用自定义Avro Schema而非Schema Registry托管的，可直接将schema字符串传入from_avro函数。

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