要使用Kafka协议中的EventHubs，您只需正确配置Kafka选项。您需要：

• 我们需要获取Shared Access Signatures (SAS)以向事件中心主题进行身份验证-它应类似于Endpoint=sb://<....>.windows.net/;?...，并将用作密码。出于安全原因，建议将其放入Databricks机密作用域（使用您的实际值更新变量secret_scope和secret_name）。
• 我们需要构造正确的字符串（eh_sasl变量）用于SASL（简单身份验证和安全层）身份验证-我们使用静态值$ConnectionString作为用户名，和事件中心SAS用作密码。SASL字符串在数据块上看起来有点不同-它应该以kafkashaded.作为前缀，而不是org.apache.kafka.common.security.plain.PlainLoginModule...，因为原始Java包被着色以避免与其他包冲突。
• 您需要提供事件中心名称空间&主题的名称，以便在eh_namespace_name和topic_name变量中从中读取数据。

secret_scope = "scope"
secret_name = "eventhub_sas"
topic_name = "topic1"
eh_namespace_name = "<eh-ns-name>"
readConnectionString = dbutils.secrets.get(secret_scope, secret_name)
eh_sasl = 'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule' \
    + f' required username="$ConnectionString" password="{readConnectionString}";'
bootstrap_servers = f"{eh_namespace_name}.servicebus.windows.net:9093"
kafka_options = {
     "kafka.bootstrap.servers": bootstrap_servers,
     "kafka.sasl.mechanism": "PLAIN",
     "kafka.security.protocol": "SASL_SSL",
     "kafka.request.timeout.ms": "60000",
     "kafka.session.timeout.ms": "30000",
     "startingOffsets": "earliest",
     "kafka.sasl.jaas.config": eh_sasl,
     "subscribe": topic_name,
}
df = spark.readStream.format("kafka") \ 
    .options(**kafka_options).load()

使用类似的配置完成写入。

from pyspark.sql.functions import struct, to_json

# work with your dataframe
kafka_options = {
     "kafka.bootstrap.servers": bootstrap_servers,
     "kafka.sasl.mechanism": "PLAIN",
     "kafka.security.protocol": "SASL_SSL",
     "kafka.sasl.jaas.config": eh_sasl,
     "topic": topic_name,
}
df.select(to_json(struct("*")).alias("value")) \
  .write.format("kafka").options(**kafka_options).save()

有关Spark + Kafka的更多详细信息，请参阅Spark和Databricks文档。