如何使用 Azure 事件中心将消息从 Databricks 流式传输到 Kafka 客户端答案

【问题标题】：How to stream messages from Databricks to a Kafka client using Azure Event Hubs如何使用 Azure 事件中心将消息从 Databricks 流式传输到 Kafka 客户端
【发布时间】：2023-01-27 19:53:07
【问题描述】：

我有一个从 Kafka 队列读取并写入 DWH 的进程。 kafka 队列当前正在从 Java 应用程序接收数据，该应用程序从本地存储读取数据并写入 Kafka 队列。

我们需要实现以下内容：

用 Azure 存储帐户替换本地存储（完成）
用 Azure 事件中心替换 Kafka 队列
用 Databricks 简单作业替换 java 应用程序，该作业使用 Azure DataLake 中的自动加载器执行 readStream，并写入 Azure 事件中心
约束：kafka客户端消费者不能改变，而不是它的连接字符串。

现在，好消息是 Azure 事件中心是 Kafka 兼容的（让我们考虑每条消息的 json 主体小于 10Kb），所以我的问题是如何配置这个架构。进一步来说：
- A) 应如何将 Azure EH 配置为与其消费者兼容 kafka？
- B) 我是否也应该使用来自 Databricks 的 kafka 协议来发送消息，或者我是否可以将它用于 Azure 事件中心，相信它通过 Kafka 接口向消费者公开自己，并通过事件中心接口向发送者公开这一事实？
- C) 哪里可以从consumer那里获取要使用的kafka endpoint，除了新的connection string还需要注意什么？在监听策略中，我找到了主键、连接字符串和 SAS 策略 ARM ID，但我不确定如何将它们转换为 Kafka enpoint

【问题讨论】：

标签： apache-kafka azure-databricks azure-eventhub

【解决方案1】：

要使用来自 Kafka 协议的 EventHubs，你只需要正确配置 Kafka 选项。您需要以下内容：

我们需要获取 Shared Access Signatures (SAS) 以对事件中心主题进行身份验证 - 它应该类似于 Endpoint=sb://<....>.windows.net/;?... 并将用作密码。出于安全原因，建议将其放入 Databricks 秘密范围（使用您的实际值更新变量 secret_scope 和 secret_name）。
我们需要为 SASL（简单身份验证和安全层）身份验证形成正确的字符串（eh_sasl 变量）- 作为用户名，我们使用静态值 $ConnectionString，事件中心 SAS 用作密码。 SASL 字符串在 Databricks 上看起来有点不同 - 而不是 org.apache.kafka.common.security.plain.PlainLoginModule... 它应该以 kafkashaded. 为前缀，因为原始 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()

写作是用类似的配置完成的。在 Spark 和 Databricks 文档中查看有关 Spark + Kafka 的更多详细信息。

【讨论】：