Databricks Notebook通过IMS Token认证连接Kafka主题的方法

正确的SASL配置格式 #

针对IMS token认证的Kafka主题，需要使用SASL_OAUTHBEARER认证机制，以下是完整的Databricks Notebook实现代码及关键配置说明：

完整代码示例 #

from pyspark.sql import SparkSession

# Databricks中可直接使用内置spark对象，无需手动初始化
spark = SparkSession.builder.appName("IMS-Kafka-Producer").getOrCreate()

# 核心配置参数
KAFKA_BROKERS = "your-kafka-broker-list:9093"
TARGET_TOPIC = "topic_name"
CLIENT_ID = "your-registered-client-id"
IMS_ACCESS_TOKEN = "valid-ims-access-token"  # 建议动态获取，避免硬编码过期token

# Kafka认证与连接配置
kafka_producer_config = {
    "kafka.bootstrap.servers": KAFKA_BROKERS,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "OAUTHBEARER",
    "kafka.sasl.jaas.config": f"""org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required
        clientId="{CLIENT_ID}"
        oauth.token="{IMS_ACCESS_TOKEN}";""",
    "kafka.sasl.login.callback.handler.class": "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler"
}

# 构造测试数据并发送
test_data = spark.createDataFrame([("sample-message-1",), ("sample-message-2",)], ["value"])
test_data.selectExpr("CAST(value AS STRING)") \
    .write \
    .format("kafka") \
    .option("topic", TARGET_TOPIC) \
    .options(**kafka_producer_config) \
    .save()

关键配置说明 #

• SASL机制与协议：必须指定security.protocol为SASL_SSL，sasl.mechanism为OAUTHBEARER，这是IMS token对应的标准认证方式。
• JAAS配置：sasl.jaas.config中通过clientId指定你的客户端ID，oauth.token传入有效的IMS访问令牌，注意格式中的引号和分号不能遗漏。
• Token动态刷新：IMS令牌有有效期，硬编码会导致后续认证失败，建议在Notebook中添加代码调用IMS的token接口自动刷新（例如用requests库发起POST请求获取最新token）。

错误排查建议 #

你遇到的TimeoutException本质是认证失败，Broker拒绝了元数据请求，导致无法识别主题：

• 验证IMS令牌的权限：确保该token拥有目标主题的生产权限。
• 检查配置语法：JAAS配置中的字符串引号、分号必须正确，避免语法错误导致认证失败。
• 网络连通性：确认Databricks集群能访问Kafka Broker的9093端口（SSL端口）。
• 依赖兼容性：确保Spark版本对应的Kafka客户端支持OAUTHBEARER机制，若集群版本过旧，可升级集群或添加兼容的Kafka客户端依赖。

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