PySpark读取Azure EventHubs持续运行后性能下降等问题咨询

我有一段PySpark代码，用于从4个Azure EventHubs读取数据，需要每分钟执行一次。代码以Databricks作业形式运行，集群配置为3个节点（16GB内存、4核）且保持持续运行状态。但持续执行1.5小时后性能下降，原本应1分钟内完成的作业耗时逐渐增加。

读取数据的代码如下：

df = spark.read.format("eventhubs") \
                .options(**eventhub_configuration) \
                .load() \
                .withColumn("eh_ky",f.lit(key)) \
                .persist()

上述代码通过循环读取所有EventHubs的数据并进行Union得到单个DataFrame，之后执行简单转换操作，最后通过更新JSON文件保存偏移量（未使用Spark Checkpointing）。

我尝试在集群环境变量中配置G1GC垃圾收集器，但未得到改善，现咨询以下问题：

1. 如何解决持续执行1.5小时后的性能下降问题？
2. 即使数据量很小，新增EventHubs作为数据源仍会增加执行时间，该如何解决？
3. 当出现连接问题时，EventHubs读取操作需等待5分钟才会失败，如何将超时时间缩短至数秒？

1. 解决长期运行后的性能下降问题 #

• 替换自定义偏移量存储为Spark Checkpointing：手动更新JSON文件保存偏移量易引发元数据泄露或重复读取，改用Spark内置的Checkpointing机制，它能自动管理偏移量并清理无效状态。读取EventHubs时配置checkpointLocation参数，路径需指向DBFS或Azure持久化存储目录。
• 移除不必要的persist()调用：若后续未重复使用单个EventHub的DataFrame，persist()会占用内存且无法及时释放，直接移除该调用，让Spark自动优化内存分配。
• 定期清理集群缓存：持续运行的集群会累积临时缓存，每次作业结束时调用spark.catalog.clearCache()清理，或设置集群每2小时自动重启一次，重置集群状态。
• 校验偏移量读取逻辑：确保每次读取正确加载上次的偏移量，避免重复读取历史数据导致数据量累积，加重集群负载。

2. 解决新增EventHubs后的执行时间增加问题 #

• 并行读取多个EventHubs：不要用循环逐个读取再Union，改用Spark并行读取方式。将所有EventHub的配置放入列表，通过sc.parallelize并行处理每个EventHub的读取任务，再一次性合并结果；或直接创建多个DataFrame后用reduce(lambda df1, df2: df1.union(df2))合并，减少执行计划复杂度。
• 优化分区与读取量：在eventhub_configuration中设置maxEventsPerTrigger控制单次读取的数据量，同时调整spark.sql.shuffle.partitions为节点数×核数×2的合理值，避免小数据量下的分区冗余开销。
• 避免重复Union操作：循环Union会生成嵌套的执行计划，改用一次性合并所有DataFrame的方式，降低Spark的计划解析开销。

3. 缩短EventHubs连接超时时间 #

• 配置EventHubs客户端超时参数：在eventhub_configuration中添加ISO 8601格式的超时参数，示例如下：

  eventhub_configuration.update({
      "connectionTimeout": "PT5S",  # 连接超时5秒
      "receiveTimeout": "PT5S"      # 数据接收超时5秒
  })
  

• 设置Databricks作业级超时：在作业配置中设置任务超时时间（如10秒），避免单个读取任务阻塞整个作业流程。

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