如何在Databricks Unity Catalog中构建青铜层每日分区表并处理全量与增量加载

1. 创建包含每日分区的bronze_customer表 #

你的ADLS路径采用标准Hive分区格式（year=YYYY/month=MM/day=DD），直接利用Unity Catalog创建外部表绑定该路径即可自动识别分区，无需手动维护分区信息。

核心实现思路： #

• 使用development.adventureworks.bronze_customer作为表的完整限定名（遵循Unity Catalog层级规则）
• 指定表的LOCATION到bronze容器下的adventureworks根路径，自动递归识别所有子分区
• 显式定义分区列year、month、day为整数类型，匹配路径中的分区键格式
• Bronze层仅做数据落地，保留原始CSV的所有字段，不做任何清洗转换

创建表代码： #

CREATE TABLE IF NOT EXISTS development.adventureworks.bronze_customer (
  customer_id INT,
  first_name STRING,
  last_name STRING,
  email STRING,
  phone STRING,
  address STRING,
  create_date TIMESTAMP,
  last_update_date TIMESTAMP
)
PARTITIONED BY (year INT, month INT, day INT)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
STORED AS TEXTFILE
LOCATION 'abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/'
TBLPROPERTIES (
  'skip.header.line.count' = '1', -- 跳过CSV表头行
  'spark.sql.sources.partitionColumnTypeInference.enabled' = 'false' -- 禁用自动分区类型推断，避免分区列变为字符串
);

初始化完成后，执行MSCK REPAIR TABLE development.adventureworks.bronze_customer;即可加载ADLS上已存在的所有分区数据。

2. 全量加载与增量加载的推荐方案 #

全量加载（每日替换所有数据） #

适用于需要每日覆盖全量历史数据的场景，推荐两种高效实现方式：

• 方式1：TRUNCATE + 全量插入
  先清空表的所有数据和分区，再重新读取ADLS上的所有文件加载，适合数据量较小的场景：

  TRUNCATE TABLE development.adventureworks.bronze_customer;
  INSERT INTO development.adventureworks.bronze_customer
  SELECT *, 
         regexp_extract(_metadata.file_path, 'year=(\\d+)', 1)::INT AS year,
         regexp_extract(_metadata.file_path, 'month=(\\d+)', 1)::INT AS month,
         regexp_extract(_metadata.file_path, 'day=(\\d+)', 1)::INT AS day
  FROM csv.`abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/`;
  

• 方式2：INSERT OVERWRITE 全量覆盖
  直接覆盖整个表，保留分区结构，执行效率更高：

  INSERT OVERWRITE TABLE development.adventureworks.bronze_customer
  SELECT *, 
         regexp_extract(_metadata.file_path, 'year=(\\d+)', 1)::INT AS year,
         regexp_extract(_metadata.file_path, 'month=(\\d+)', 1)::INT AS month,
         regexp_extract(_metadata.file_path, 'day=(\\d+)', 1)::INT AS day
  FROM csv.`abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/`;
  

增量加载（仅追加新/变更数据） #

针对ADF每日上传的新分区文件，推荐两种Databricks原生工具：

• 方案1：COPY INTO（推荐，简单高效）
  自动追踪已加载的文件，仅处理新上传的文件，无需手动管理加载状态：

  COPY INTO development.adventureworks.bronze_customer
  FROM (
    SELECT *, 
           regexp_extract(_metadata.file_path, 'year=(\\d+)', 1)::INT AS year,
           regexp_extract(_metadata.file_path, 'month=(\\d+)', 1)::INT AS month,
           regexp_extract(_metadata.file_path, 'day=(\\d+)', 1)::INT AS day
    FROM 'abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/'
  )
  FILEFORMAT = CSV
  OPTIONS ('header' = 'true')
  COPY_OPTIONS ('mergeSchema' = 'true'); -- 支持模式演化，新增字段自动识别
  

• 方案2：Auto Loader（适合大量小文件/模式演化）
  利用云原生文件通知机制，实时或批量发现新文件，支持自动分区和模式演化：

  from pyspark.sql.streaming import Trigger
  
  df = (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "csv")
        .option("cloudFiles.schemaLocation", "abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/_schema/")
        .option("header", "true")
        .load("abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/")
        .withColumn("year", regexp_extract("_metadata.file_path", "year=(\\d+)", 1).cast("int"))
        .withColumn("month", regexp_extract("_metadata.file_path", "month=(\\d+)", 1).cast("int"))
        .withColumn("day", regexp_extract("_metadata.file_path", "day=(\\d+)", 1).cast("int"))
       )
  
  (df.writeStream
   .option("checkpointLocation", "abfss://bronze@<your-adls-account-name>.dfs.core.windows.net/adventureworks/_checkpoint/")
   .trigger(Trigger.Once()) -- 每日批量执行一次，替换为Trigger.ProcessingTime("1 hour")可实时处理
   .toTable("development.adventureworks.bronze_customer")
  )
  

3. 最佳实践总结 #

• Bronze层核心原则：保留原始数据格式和所有字段，仅做数据落地和分区绑定，不做清洗转换
• 分区优化：使用year/month/day三级分区，减少查询时的数据扫描范围，提升查询效率
• 小文件治理：定期执行OPTIMIZE development.adventureworks.bronze_customer ZORDER BY (customer_id);合并小文件，VACUUM development.adventureworks.bronze_customer RETAIN 7 DAYS;清理旧数据文件
• 权限控制：通过Unity Catalog给不同角色分配bronze表的读取/写入权限，比如给ETL服务账号分配写入权限，给分析人员分配读取权限
• 状态管理：使用COPY INTO或Auto Loader的内置状态追踪，避免重复加载文件
• 日志记录：在加载过程中记录文件路径、加载时间、行数等信息，便于问题排查

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