Databricks Notebook读取Confluent Kafka流数据遇初始化停滞求助

一、正确的Confluent Kafka连接配置（SASL_SSL认证） #

Confluent Cloud的Kafka集群默认强制SASL_SSL认证，必须配置完整的JAAS参数才能正常连接。你之前的问题核心是凭证配置不完整，以下是可直接复用的正确代码：

kafka_df = (
    spark
    .readStream
    .format("kafka")
    # 替换为Confluent控制台Cluster Settings里的完整Bootstrap地址（带9092端口）
    .option("kafka.bootstrap.servers", "pkc-xxx.us-west2.gcp.confluent.cloud:9092")
    .option("subscribe", "topic_0")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.mechanism", "PLAIN")
    # 替换为你的Confluent API Key和对应的Secret（不是账号密码，是专门的Kafka访问凭证）
    .option("kafka.sasl.jaas.config", 
            'org.apache.kafka.common.security.plain.PlainLoginModule required username="你的API_KEY" password="你的API_SECRET";')
    .option("kafka.request.timeout.ms", "60000")
    .option("kafka.session.timeout.ms", "30000")
    .load()
)

display(kafka_df)

注意点：

• JAAS配置的引号要外层用单引号、内层用双引号，避免Python语法错误
• API Key和Secret需要在Confluent Cloud的API Keys页面创建，且要给该凭证分配目标Topic的读取权限

二、排查连接失败的常见原因 #

1. 网络连通性：检查Databricks集群是否能访问Confluent的Bootstrap服务器。可以在笔记本中执行nc -zv <bootstrap-server> 9092测试（需要先安装netcat：%sh apt-get install -y netcat），如果不通，要排查VPC防火墙规则是否放行9092端口的出站流量
2. 凭证有效性：确认API Key/Secret没有复制错误（比如多了空格），且该凭证已被授权读取目标Topic
3. Topic正确性：确认topic_0在Confluent集群中存在，且拼写完全匹配（Kafka Topic大小写敏感）
4. 批处理读取的特殊情况：用spark.read测试时，如果Topic中没有数据，命令会一直等待。可以添加.option("startingOffsets", "earliest")或.option("endingOffsets", "latest")限定读取范围，避免无限阻塞
5. 版本兼容性：确保Databricks预装的Spark Kafka连接器版本与Confluent Kafka版本匹配（Databricks官方集群一般会预装兼容版本，自定义集群需额外确认）

三、Delta Live Tables适配说明 #

如果后续要构建Delta Live Tables，配置逻辑一致，可直接在DLT代码中嵌入配置：

import dlt

@dlt.table
def kafka_raw_table():
    return (
        spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "<你的Bootstrap地址>")
        .option("subscribe", "topic_0")
        .option("kafka.security.protocol", "SASL_SSL")
        .option("kafka.sasl.mechanism", "PLAIN")
        .option("kafka.sasl.jaas.config", 
                'org.apache.kafka.common.security.plain.PlainLoginModule required username="API_KEY" password="API_SECRET";')
        .load()
    )

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