使用 EventHub 的 Delta 实时表答案

【问题标题】：Delta Live Tables with EventHub使用 EventHub 的 Delta 实时表
【发布时间】：2022-07-31 03:43:42
【问题描述】：

我正在尝试使用 delta 实时表从 eventthub 创建流，但我在安装库时遇到问题。是否可以使用 sh /pip 使用 Delta Live 表安装 maven 库？

我想安装 com.microsoft.azure:azure-eventhubs-spark_2.12:2.3.17

https://docs.microsoft.com/pl-pl/azure/databricks/spark/latest/structured-streaming/streaming-event-hubs

【问题讨论】：

目前 delta live tables 框架不支持集群初始化脚本，我添加了这条评论以防止其他用户尝试使用它来解决这个问题。

标签： pyspark databricks azure-eventhub delta-live-tables

【解决方案1】：

目前无法为 Delta Live Tables 使用外部连接器/Java 库。但是对于 EventHubs 有一个解决方法 - 您可以使用内置的 Kafka 连接器连接到 EventHubs - 您只需要指定正确的选项，如documentation 中所述：

@dlt.table
def eventhubs():
  readConnectionString="Endpoint=sb://<....>.windows.net/;?.."
  eh_sasl = f'kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="{readConnectionString}";'
  kafka_options = {
     "kafka.bootstrap.servers": "<eh-ns-name>.servicebus.windows.net:9093",
     "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>",
  }
  return spark.readStream.format("kafka") \ 
    .options(**kafka_options).load()

【讨论】：