Databricks中Kafka流处理的Azure Key Vault凭证配置及运行问题

1. 在Structured Streaming中配置Kafka SASL凭证 #

Spark Structured Streaming连接SASL认证的Kafka时，需通过option添加安全相关配置，直接复用你从Azure Key Vault获取的凭证即可：

# 从AKV获取凭证
username = dbutils.secrets.get(scope = "DATAAI-CEC", key = "dai-kafka-cec-api-key")
pwd = dbutils.secrets.get(scope = "DATAAI-CEC", key = "dai-kafka-cec-secret")

from pyspark.sql.types import *
from pyspark.sql.functions import from_json, col

# 定义消息Schema
schema = StructType() \
  .add("EventHeader", StructType() \
    .add("UUID", StringType()) \
    .add("APPLICATION_ID", StringType())
    .add("FORMAT", StringType())) \
  .add("EmissionReportMessage", StructType() \
    .add("reportId", StringType()) \
    .add("startDate", StringType()) \
    .add("endDate", StringType()) \
    .add("unitOfMeasure", StringType()) \
    .add("reportLanguage", StringType()) \
    .add("companies", ArrayType(StructType([StructField("ccid", StringType(), True)]))))

# 配置Kafka安全参数
parsed_kafka = spark \
                .readStream \
                .format("kafka") \
                .option("kafka.bootstrap.servers", "SERVER:PORT") \
                .option("subscribe", "TOPIC") \
                .option("startingOffsets", "earliest") \
                # 添加SASL认证配置
                .option("kafka.security.protocol", "SASL_SSL") \
                .option("kafka.sasl.mechanism", "PLAIN") \
                .option("kafka.sasl.jaas.config", 
                        f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{username}" password="{pwd}";') \
                .load()\
                .select(from_json(col("value").cast("string"), schema).alias("kafka_parsed_value"))

注意：sasl.jaas.config格式固定，需用字符串格式化注入用户名和密码，确保引号转义正确。

2. 解决display无输出的问题 #

执行display(parsed_kafka)后无结果，可从以下方向排查：

• 检查Kafka Topic数据情况：若startingOffsets设为earliest但Topic无历史消息、也无新消息写入，自然不会有输出。可先往Topic发送测试消息，再观察display结果。
• 验证Schema匹配度：若消息JSON结构与定义的schema不匹配，from_json会返回null，导致输出为空。可临时去掉解析逻辑，直接查看原始消息：

  raw_kafka = spark \
                .readStream \
                .format("kafka") \
                .option("kafka.bootstrap.servers", "SERVER:PORT") \
                .option("subscribe", "TOPIC") \
                .option("startingOffsets", "earliest") \
                .option("kafka.security.protocol", "SASL_SSL") \
                .option("kafka.sasl.mechanism", "PLAIN") \
                .option("kafka.sasl.jaas.config", f'org.apache.kafka.common.security.plain.PlainLoginModule required username="{username}" password="{pwd}";') \
                .load()\
                .select(col("value").cast("string").alias("raw_message"))
  
  display(raw_kafka)
  

  查看原始消息结构后，调整schema使其匹配。
• 确认Kafka连接有效性：检查bootstrap.servers地址是否正确、凭证是否有效（可复用之前的kafka-python代码测试连接拉取能力），确保Databricks集群能访问Kafka服务器端口。
• 查看流查询状态：在Databricks的「Jobs」页面查看流查询状态，是否存在认证失败、网络不通等报错信息。

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