Databricks使用SASL_SSL&SCRAM-SHA-512连接Kafka失败求助

可能的失败原因 #

• 版本兼容性问题：Databricks集群的Spark Kafka连接器版本过低，可能不支持SCRAM-SHA-512认证机制，需确认连接器版本与目标Kafka集群版本匹配（Kafka 0.10.2及以上版本支持SCRAM-SHA-512）。
• 认证参数配置错误：部分Spark Kafka连接器版本不支持单独通过kafka.sasl.username和kafka.sasl.password传递凭证，需使用完整的JAAS配置字符串。
• SSL证书信任问题：云端Kafka集群的SSL证书未被Databricks集群信任，导致SSL握手失败。
• 权限或凭证无效：用户名/密码错误，或该用户未被授予目标Topic的消费权限。
• 网络连通性问题：Databricks集群未打通到Kafka集群9094端口的网络通路，无法建立连接。

修正后的可用示例代码 #

Databricks环境中无需指定master("local[*]")（集群资源由平台自动管理），同时需通过JAAS配置传递SCRAM认证信息：

from pyspark.sql import SparkSession

# Databricks内置spark变量可直接使用，此处保留构建代码仅作示例参考
spark = SparkSession.builder \
    .appName("KafkaStreaming") \
    .getOrCreate()

# 构建SCRAM-SHA-512的JAAS配置字符串
jaas_config = """org.apache.kafka.common.security.scram.ScramLoginModule required
username="abc"
password="abc";"""

kafka_stream_df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "moped.srvs.cloudkafka.com:9094") \
    .option("kafka.security.protocol", "SASL_SSL") \
    .option("kafka.sasl.mechanism", "SCRAM-SHA-512") \
    .option("kafka.sasl.jaas.config", jaas_config) \
    # 若遇SSL证书信任问题，可临时添加以下参数（生产环境建议使用信任证书而非跳过验证）
    # .option("kafka.ssl.endpoint.identification.algorithm", "")
    # .option("kafka.ssl.truststore.type", "PEM")
    .option("startingoffsets", "earliest") \
    .option("subscribe", "abc-default") \
    .load()

# 解析Kafka消息的key、value为可读字符串（按需扩展处理逻辑）
parsed_stream_df = kafka_stream_df.selectExpr(
    "CAST(key AS STRING)",
    "CAST(value AS STRING)",
    "topic",
    "partition",
    "offset"
)

# 输出到控制台查看消费结果
query = parsed_stream_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .option("truncate", False) \
    .start()

query.awaitTermination()

额外排查建议 #

• 若使用自定义SSL证书，需将证书上传至DBFS，并配置kafka.ssl.truststore.location和kafka.ssl.truststore.password参数。
• 检查Databricks集群的安全组/网络规则，确保允许访问Kafka集群的9094端口。
• 通过查看Databricks集群的Spark驱动日志或Executor日志，获取更详细的错误信息以定位具体问题。

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