如何用PySpark 2.2从Confluent Kafka 4.0.0读取Avro格式Kafka主题数据

我来帮你完善这个读取Kafka Avro数据的PySpark Streaming程序，结合你使用的Confluent Kafka 4.0.0环境，下面是完整的实现方案：

1. 先确认依赖包 #

运行程序前，你需要确保Spark环境包含以下依赖（可以通过spark-submit的--packages参数引入）：

• Spark Streaming Kafka整合包：org.apache.spark:spark-streaming-kafka-0-10_2.11:2.3.0（Confluent 4.0.0对应Kafka 1.0.0，Spark 2.3.x版本兼容性较好）
• Confluent Avro相关包：io.confluent:kafka-avro-serializer:4.0.0

2. 完整的PySpark Streaming代码 #

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils
from confluent_kafka.avro.cached_schema_registry_client import CachedSchemaRegistryClient
from confluent_kafka.avro.serializer.message_serializer import MessageSerializer

# 初始化Schema Registry客户端
schema_registry_url = "http://your-schema-registry-host:8081"  # 替换为你的Schema Registry地址
schema_registry_client = CachedSchemaRegistryClient({'url': schema_registry_url})
message_serializer = MessageSerializer(schema_registry_client)

# 初始化Spark上下文和Streaming上下文
sc = SparkContext(appName="KafkaAvroStreamingReader")
ssc = StreamingContext(sc, 5)  # 设置5秒的批处理间隔，可根据需求调整

# Kafka配置参数
kafka_params = {
    "bootstrap.servers": "your-kafka-broker:9092",  # 替换为你的Kafka Broker地址
    "group.id": "spark-avro-consumer-group",
    "auto.offset.reset": "latest"  # 根据业务需求选择earliest或latest
}

# 要读取的目标Kafka主题
topics = ["your-source-topic"]  # 替换为你从SQL Server抽取数据的主题

# 创建Kafka Direct DStream
kafka_stream = KafkaUtils.createDirectStream(ssc, topics, kafka_params)

# 定义Avro反序列化函数
def deserialize_avro(value):
    if value is None:
        return None
    # 使用Confluent的序列化工具解码Avro数据
    return message_serializer.decode_message(value)

# 定义每个批次RDD的处理逻辑
def process_rdd(rdd):
    if not rdd.isEmpty():
        # 对每条消息的value进行反序列化
        avro_records = rdd.map(lambda x: deserialize_avro(x[1]))
        # 这里可以添加你的业务逻辑，比如打印数据、写入数仓/数据库等
        avro_records.foreach(lambda record: print(f"解析后的Avro数据: {record}"))

# 将处理逻辑应用到DStream
kafka_stream.foreachRDD(process_rdd)

# 启动Streaming作业并等待终止
ssc.start()
ssc.awaitTermination()

3. 关键部分解释 #

• Schema Registry客户端：CachedSchemaRegistryClient会缓存已获取的Schema，避免重复请求Schema Registry，有效提升反序列化性能。
• MessageSerializer：Confluent官方提供的工具，能自动从Schema Registry拉取对应版本的Schema，完成Avro数据的解码，无需手动指定Schema。
• Direct DStream模式：相比Receiver模式，直接连接Kafka Broker获取数据，延迟更低且能保证Exactly-Once语义，更适合生产环境。

4. 运行注意事项 #

• 替换代码中所有带有your-前缀的配置项为你的实际环境信息。
• 如果在集群环境运行，确保所有Worker节点都能访问Kafka Broker和Schema Registry服务。
• 严格注意版本匹配：Confluent 4.0.0对应Kafka 1.0.0，Spark版本建议用2.3.x，避免版本不兼容导致的异常。

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