STREAMING_CONNECT_SERIALIZATION_ERROR：Databricks Notebook向Azure EventHub发流数据报错

错误信息 #

[STREAMING_CONNECT_SERIALIZATION_ERROR] 无法序列化foreachBatch函数。若你在函数外部访问了Spark会话、DataFrame或任何包含Spark会话的对象，请注意Spark Connect不允许此类操作。对于foreachBatch，请通过该函数的第一个参数df的df.sparkSession来访问Spark会话；对于StreamingQueryListener，请使用self.spark访问。详情请查阅PySpark中foreachBatch和StreamingQueryListener的文档。

问题分析 #

出现该错误的核心原因：

• process_batch函数依赖了全局变量last_sent_time，全局变量无法被Spark序列化并分发到执行节点
• 若fetch_weather_data或send_event函数中引用了全局的spark实例、EventHub生产者等外部对象，同样会导致序列化失败
• Spark的流处理要求foreachBatch函数必须是可序列化的，不能包含外部不可序列化的依赖

解决方案 #

关键修改点 #

• 移除全局变量，改用Spark状态存储（检查点、临时表）维护批次间的状态
• 通过batch_df.sparkSession在函数内部获取Spark会话，禁止使用全局spark实例
• 将EventHub生产者的创建逻辑移到函数内部，避免全局对象引用

修改后的代码示例 #

import datetime
from azure.eventhub import EventHubProducerClient, EventData

def fetch_weather_data(spark_session):
    # 使用传入的spark_session执行数据操作，示例返回模拟天气数据
    return {
        "temperature": 25, 
        "humidity": 60, 
        "timestamp": datetime.datetime.now().isoformat()
    }

def send_event(weather_data, conn_str, eventhub_name):
    # 在函数内部创建EventHub生产者，避免全局引用
    producer = EventHubProducerClient.from_connection_string(
        conn_str=conn_str,
        eventhub_name=eventhub_name
    )
    with producer:
        batch = producer.create_batch()
        batch.add(EventData(str(weather_data)))
        producer.send_batch(batch)

def process_batch(batch_df, batch_id):
    # 从当前批次DataFrame获取Spark会话
    spark_session = batch_df.sparkSession
    
    # 读取上次发送时间（首次运行初始化状态）
    try:
        last_sent_row = spark_session.read.table("eventhub_last_sent").collect()[0]
        last_sent_time = last_sent_row["last_sent"]
    except:
        last_sent_time = datetime.datetime.now() - datetime.timedelta(seconds=30)
    
    current_time = datetime.datetime.now()
    if (current_time - last_sent_time).total_seconds() >= 30:
        weather_data = fetch_weather_data(spark_session)
        # 替换为你的EventHub连接信息
        send_event(weather_data, "YOUR_EVENTHUB_CONN_STR", "YOUR_EVENTHUB_NAME")
        
        # 更新发送状态到临时表
        spark_session.createDataFrame(
            [(current_time,)], ["last_sent"]
        ).write.mode("overwrite").saveAsTable("eventhub_last_sent")
        
        print(f'Event Sent at {current_time}')

# 初始化流数据源
streaming_df = spark.readStream.format("rate").option("rowsPerSecond", 1).load()

# 启用检查点保证流状态一致性
query = streaming_df.writeStream.foreachBatch(process_batch)\
    .option("checkpointLocation", "/dbfs/checkpoints/eventhub_stream_checkpoint")\
    .start()

query.awaitTermination()

额外说明 #

• 检查点路径需使用DBFS或Azure Storage路径，确保流中断后能恢复状态
• 若直接使用Spark的EventHub数据源写入，可完全避免自定义foreachBatch的序列化问题，示例代码：

  streaming_df.writeStream.format("eventhubs")\
      .option("eventhubs.connectionString", "YOUR_CONN_STR")\
      .option("checkpointLocation", "/dbfs/checkpoints/eventhub_direct")\
      .start()
  

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