求Databricks Structured Streaming对接Azure Event Grid的文档与示例

Databricks Structured Streaming 没有原生的 Azure Event Grid（AEG）连接器，通常需要借助 Event Grid 的事件转发能力，将事件路由到 Databricks 支持的中间数据源后，再进行流式处理。以下是具体方案及示例：

方案1：通过 Azure Blob Storage 中转 #

1. 配置 Event Grid，将目标事件发送到指定的 Blob Storage 容器（作为事件存储端点）
2. 使用 Databricks Structured Streaming 读取 Blob Storage 中的事件文件

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, TimestampType

# 初始化Spark会话
spark = SparkSession.builder.appName("EventGridToDatabricks").getOrCreate()

# 定义Event Grid事件的Schema结构
event_grid_schema = StructType() \
    .add("id", StringType()) \
    .add("topic", StringType()) \
    .add("subject", StringType()) \
    .add("eventType", StringType()) \
    .add("eventTime", TimestampType()) \
    .add("data", StringType()) \
    .add("dataVersion", StringType()) \
    .add("metadataVersion", StringType())

# 流式读取Blob Storage中的Event Grid事件文件
stream_df = spark.readStream \
    .format("cloudFiles") \
    .option("cloudFiles.format", "json") \
    .option("cloudFiles.schemaLocation", "/dbfs/path/to/schema-storage") \
    .load("wasbs://<容器名称>@<存储账户>.blob.core.windows.net/event-grid-events/")

# 解析事件数据
parsed_df = stream_df.select(from_json(col("value"), event_grid_schema).alias("event")) \
    .select("event.*")

# 示例：输出到控制台
query = parsed_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

query.awaitTermination()

方案2：通过 Azure Event Hubs 中转 #

1. 配置 Event Grid，将事件转发到指定的 Event Hubs 实例
2. 使用 Databricks 的 Event Hubs 连接器读取流式数据

示例代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, TimestampType

spark = SparkSession.builder.appName("EventGridViaEventHubs").getOrCreate()

# 定义Event Grid事件的Schema结构
event_grid_schema = StructType() \
    .add("id", StringType()) \
    .add("topic", StringType()) \
    .add("subject", StringType()) \
    .add("eventType", StringType()) \
    .add("eventTime", TimestampType()) \
    .add("data", StringType()) \
    .add("dataVersion", StringType()) \
    .add("metadataVersion", StringType())

# Event Hubs连接配置
eh_conf = {
  "eventhubs.connectionString": "Endpoint=sb://<命名空间>.servicebus.windows.net/;SharedAccessKeyName=<密钥名称>;SharedAccessKey=<密钥值>;EntityPath=<Event Hub名称>"
}

# 流式读取Event Hubs中的Event Grid事件
stream_df = spark.readStream \
    .format("eventhubs") \
    .options(**eh_conf) \
    .load()

# 解析事件内容
parsed_df = stream_df.select(from_json(col("body").cast("string"), event_grid_schema).alias("event")) \
    .select("event.*")

# 示例：输出到Delta表
query = parsed_df.writeStream \
    .format("delta") \
    .option("checkpointLocation", "/dbfs/path/to/checkpoint-folder") \
    .start("/dbfs/path/to/delta-table")

query.awaitTermination()

关键说明 #

• Azure Event Grid 本质是事件路由服务，并非流式数据源，因此必须通过中间存储或消息服务对接 Databricks
• 中转方案选择：Blob Storage 适合批量处理、延迟要求较低的场景；Event Hubs 适合高吞吐量、低延迟的流式场景

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