如何在Bronze层处理CSV文件，避免形成Raw+Bronze+Silver+Gold冗余层级？

先明确各层核心定位 #

• Raw层：纯原始CSV冷存储，完全保留数据源原样，只做分区归档，不碰数据内容，用于溯源兜底。
• Bronze层：把Raw层CSV转成Delta格式，仅做轻量增强（加元数据），不修改原始数据结构，是结构化原始数据的入口。
• Silver层：做数据清洗、标准化，输出干净的业务可用结构化数据，是业务逻辑处理的基础层。
• Gold层：针对具体业务需求做聚合、建模，直接输出可用于分析、报表的结果数据集。

1. Raw层：原始CSV持久化 #

直接按业务/时间维度归档原始CSV，不做任何数据修改，保证原始数据的完整性。
示例（Spark批量写入）：

spark.read
  .option("header", "true")
  .option("inferSchema", "false") // 不自动推断Schema，避免破坏原始数据类型
  .csv("/path/to/source/csv_files/")
  .write
  .mode("append")
  .partitionBy("load_date") // 按加载日期分区，方便后续增量读取
  .csv("/raw_layer/user_reg_csv/")

2. Bronze层：CSV转Delta并增强元数据 #

核心是把Raw层的CSV转换成Delta格式，同时添加元数据字段用于溯源，不做数据清洗。
处理步骤：

1. 读取Raw层的分区CSV
2. 追加load_timestamp（写入Bronze的时间）、source_file（原始CSV路径）等元数据
3. 写入Delta表并开启版本控制
   示例代码：

import io.delta.tables._
import org.apache.spark.sql.functions.{current_timestamp, input_file_name}

// 读取Raw层CSV
val rawCsvDF = spark.read
  .option("header", "true")
  .option("inferSchema", "true") // 这里可推断Schema，因为Bronze需要结构化存储
  .csv("/raw_layer/user_reg_csv/")

// 添加元数据字段
val bronzeDF = rawCsvDF
  .withColumn("load_timestamp", current_timestamp())
  .withColumn("source_file", input_file_name())

// 写入Bronze层Delta表
bronzeDF.write
  .format("delta")
  .mode("append")
  .partitionBy("load_date") // 和Raw层分区一致，方便增量同步
  .save("/bronze_layer/user_reg_delta/")

// 可选：注册Delta表到元数据 Catalog
spark.sql("""
  CREATE TABLE bronze_user_reg
  USING DELTA
  LOCATION '/bronze_layer/user_reg_delta/'
""")

注意：Bronze层必须保留所有原始字段，哪怕是脏数据，只做格式转换和元数据增强。

3. Silver层：数据清洗与标准化 #

基于Bronze层的Delta表，做业务规则内的清洗，输出干净的标准化数据。
常见处理动作：

• 去重（按业务主键）
• 缺失值填充/标记
• 数据类型校准（比如字符串转日期、数值）
• 过滤无效记录（比如主键为空、格式错误的数据）
  示例代码：

import org.apache.spark.sql.functions.{to_date, col}

// 读取Bronze层Delta表
val bronzeDF = spark.read.format("delta").load("/bronze_layer/user_reg_delta/")

// 清洗处理
val silverDF = bronzeDF
  // 按用户ID去重
  .dropDuplicates("user_id")
  // 填充缺失值：年龄用0填充，邮箱用默认值
  .fillna(Map("age" -> 0, "email" -> "unknown@example.com"))
  // 转换日期格式：把字符串类型的注册时间转成日期类型
  .withColumn("register_date", to_date(col("register_time"), "yyyy-MM-dd"))
  // 过滤掉用户ID为空的无效记录
  .filter(col("user_id").isNotNull)

// 写入Silver层Delta表，开启Z-Order优化（针对高频查询字段）
silverDF.write
  .format("delta")
  .mode("append")
  .partitionBy("register_date")
  .option("zOrderBy", "user_id")
  .save("/silver_layer/cleaned_user_reg/")

4. Gold层：业务聚合与建模 #

基于Silver层的干净数据，针对具体业务需求构建聚合模型，直接输出可用于分析的结果。
示例：构建每日用户注册统计报表

import org.apache.spark.sql.functions.{countDistinct, avg, to_date}

// 读取Silver层数据
val silverDF = spark.read.format("delta").load("/silver_layer/cleaned_user_reg/")

// 按日期聚合统计
val dailyRegStatsDF = silverDF
  .groupBy(to_date(col("register_date")).alias("stat_date"))
  .agg(
    countDistinct("user_id").alias("total_reg_users"),
    avg("age").alias("avg_user_age")
  )

// 写入Gold层Delta表，用覆盖模式（每日统计需更新）
dailyRegStatsDF.write
  .format("delta")
  .mode("overwrite")
  .partitionBy("stat_date")
  .save("/gold_layer/daily_user_reg_stats/")

额外优化建议 #

• 增量同步：开启Delta的change data feed功能，实现Raw→Bronze、Bronze→Silver的增量处理，避免全量扫描提升效率。
• 数据质量监控：在Silver层加入校验逻辑（比如检查字段格式、范围），确保输出数据符合业务规则。
• Delta优化：定期对Silver/Gold层执行OPTIMIZE和VACUUM命令，压缩数据、清理历史版本，提升查询性能。

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