Databricks结构化Streaming内存与磁盘占用过高问题排查求助

1. 强制清理Spark IO缓存与检查点元数据 #

Spark的IO缓存默认会保留已处理数据的缓存，长期流运行会导致无意义累积。添加以下配置到集群或作业：

• 限制缓存内存占比：spark.memory.fraction=0.4（降低默认的0.6配额，迫使缓存自动淘汰旧数据）
• 开启定期缓存清理：spark.cleaner.periodicGC.interval=300s（每5分钟触发一次垃圾回收，清理过期缓存）
• 配置检查点与源文件自动清理：在流启动时补充参数：

  .option("checkpointLocation", "/path/to/checkpoint")
  .option("cleanSource", "delete")  # 业务允许的话，处理完的源文件自动删除
  .option("maxFilesPerTrigger", 1)  # 保持单文件批次，避免单次加载过多数据
  

2. 优化foreachBatch中的Delta Merge操作 #

每次Merge若扫描全表，会导致大量数据被缓存且无法释放，需缩小扫描范围：

• 给目标Delta表添加索引：按Merge匹配的索引列创建分区或Z-Order索引，减少扫描量：

  ALTER TABLE target_delta_table ADD PARTITION INDEX (index_column);
  -- 或Z-Order优化
  OPTIMIZE target_delta_table ZORDER BY (index_column);
  

• Merge时限定批次数据范围：仅匹配当前批次涉及的索引列值，避免全表扫描：

  def merge_batch(batch_df, batch_id):
      # 提取当前批次的索引列唯一值
      index_values = [row[0] for row in batch_df.select("index_column").distinct().collect()]
      index_filter = f"index_column IN ({','.join([f"'{v}'" for v in index_values])})"
      
      batch_df.createOrReplaceTempView("batch_data")
      batch_df.sparkSession.sql(f"""
          MERGE INTO target_delta_table t
          USING batch_data b
          ON t.index_column = b.index_column AND {index_filter}
          WHEN MATCHED THEN UPDATE SET col1 = b.col1, col2 = b.col2
          WHEN NOT MATCHED THEN INSERT *
      """)
  

3. 调整Auto Loader的文件跟踪缓存 #

使用Azure队列通知的Auto Loader，需避免重复跟踪文件导致的缓存累积：

• 启用文件列表缓存过期：spark.databricks.cloudFiles.fileListingCache.enabled=true和spark.databricks.cloudFiles.fileListingCache.expirationIntervalMs=3600000（每小时过期文件列表缓存）
• 关闭源文件缓存：spark.databricks.cloudFiles.cache.enabled=false（若无需重复访问源文件，直接关闭缓存）

4. 集群与Delta表的磁盘优化 #

• 清理临时文件：添加集群初始化脚本定期清理临时目录：

  # 初始化脚本示例
  sudo find /dbfs/tmp -type f -mtime +1 -delete
  sudo find /local_disk0/tmp -type f -mtime +1 -delete
  

• 限制Delta表版本保留：避免旧版本占用过多磁盘：

  ALTER TABLE target_delta_table SET TBLPROPERTIES (
      'delta.logRetentionDuration' = '7 days',
      'delta.deletedFileRetentionDuration' = '1 day'
  );
  VACUUM target_delta_table RETAIN 168 HOURS;  # 清理7天前的旧版本文件
  

5. 流处理批次控制 #

• 保持trigger(processingTime="10 minutes")的同时，确保maxFilesPerTrigger=1的配置生效，避免单批次加载过量数据
• 所有批次内的转换（新增列、去重、分组）仅基于当前批次数据，禁止执行全局聚合或全表扫描操作

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