Databricks流转批处理湖仓三层加工及事实表处理最佳实践咨询

Bronze到Silver层数据追加优化 #

• 优先用增量变更捕获（CDC）模式代替全量MERGE：如果源数据有明确的更新时间戳或者CDC标记（比如_op标识），每次微批次只处理当前批次新增/变更的记录，直接append到silver表，不需要全表匹配。如果必须要去重，在微批次内先做本地去重，再写入，避免跨批次全表扫描。
• 必用MERGE的场景下做性能优化：
  • 给MERGE的匹配条件字段加Delta Z-Order索引，减少扫描的数据量
  • 加入分区剪枝条件，比如silver_table.p_date >= date_add(current_date(), -7)，限定只扫描近期可能有变更的分区，不要扫描全表
  • 开启Delta的merge.schemaEvolution.enabled和autoOptimize.autoCompact配置，避免小文件过多拖慢性能

Silver到Gold层事实表加工优化 #

• 采用微批次快照+增量聚合模式：事实表大多是累加型指标，每次微批次只针对当前批次的silver数据做聚合，再和历史gold表的聚合结果做合并，不要每次都全量重新计算所有历史数据
• 对于需要关联多维度表的事实表加工，开启Delta Lake的动态分区修剪和广播join优化，小维度表直接广播到大表的每个节点，避免shuffle开销
• 如果是周期性调度的批任务和流任务共用一套逻辑，直接复用相同的DataFrame加工代码，只需要把输入源从readStream换成read即可，Databricks原生支持流批代码逻辑复用

当前代码优化建议 #

你当前使用的trigger(once=True)属于触发式流处理，适配ADF调度的场景可以做如下调整：

# 新增流批统一配置，开启小文件自动合并
spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", "true")
spark.conf.set("spark.databricks.delta.autoCompact.enabled", "true")

# Autoloader读取逻辑保持不变
raw_df = (spark
          .readStream.format("cloudFiles")
          .options(**cloudfile)
          .load(raw_path)
         )

# 写入bronze表新增幂等、性能优化配置
autoloader_query = (raw_df.writeStream
                 .format("delta")
                 .trigger(availableNow=True) # 替换once=True，支持多批次并行处理新增文件，性能更高
                 .option("checkpointLocation",checkpoint_path)
                 .option("txnVersion", "1") # 开启事务保证幂等，避免ADF重试导致重复写入
                 .option("txnAppId", "bronze_ingestion")
                 .partitionBy("p_date","p_hour")
                 .table("raw_table")
                )

autoloader_query.awaitTermination()

autoloader_query.recentProgress

流批统一架构通用规范 #

• 所有层的表都采用Delta格式存储，统一事务、索引、快照能力
• checkpoint路径单独存放，不要和数据表路径混用，避免意外删除导致流任务状态丢失
• 微批次的处理大小根据数据量调整，用maxFilesPerTrigger参数控制每次处理的文件数，避免批次过大OOM或者过小产生太多小文件

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