数据分区逻辑设计

以下为 dwd_clicks 的分区与生命周期管理逻辑设计，满足 dt 月分区、region 二级分区、保留周期 365 天的要求。内容侧重可落地的 DDL、写入与清理策略，以及边界条件与注意事项。

一、分区设计与目录布局

• 分区列
  • dt: string，格式为 yyyyMM（按月分区）
  • region: string，二级分区，建议使用小写区域码（例如 cn、us、eu）。未知时统一填充 unknown。
• HDFS/S3 路径结构
  • .../warehouse/dwd.db/dwd_clicks/dt=YYYYMM/region=xx/
• 推荐存储
  • 列式格式（Parquet/ORC，压缩 snappy/zlib），避免小文件。
• Hive 表定义（示例）
  • 建议 External 表，避免误删数据；严格 schema 以匹配下游消费。
  • 示例字段按常见点击流抽象给出，可根据实际补充。

示例 DDL（Hive/Spark SQL 兼容） create database if not exists dwd; create external table if not exists dwd.dwd_clicks ( click_id string, user_id string, event_time timestamp, page_id string, referer string, device_type string, os string, app_version string, ip string, -- 其他业务列... etl_load_time timestamp ) partitioned by (dt string, region string) stored as parquet location 'hdfs:///warehouse/dwd.db/dwd_clicks' tblproperties ( 'parquet.compression'='SNAPPY' );

二、分区派生与写入逻辑

• dt 派生规则

  • dt = date_format(event_time, 'yyyyMM')（事件发生时间按月归档）
  • 若原始数据只有 event_date，确保为 UTC 或统一时区后再派生，否则会跨月偏移。

• region 派生规则

  • 优先使用可信字段（如请求头、用户配置）；无则使用 IP -> Geo 映射，失败则填 unknown。
  • 统一为 lower(region) 并做白名单/字典校验，避免产生脏分区。

• 动态分区写入（Hive）

  • 设置 set hive.exec.dynamic.partition=true; set hive.exec.dynamic.partition.mode=nonstrict; set hive.exec.max.dynamic.partitions.pernode=2000; set hive.exec.max.dynamic.partitions=20000;
  • 写入语句（追加） insert into table dwd.dwd_clicks partition (dt, region) select click_id, user_id, event_time, page_id, referer, device_type, os, app_version, ip, current_timestamp() as etl_load_time, date_format(event_time, 'yyyyMM') as dt, coalesce(lower(region), 'unknown') as region from ods_clicks_incremental; -- 按需替换数据源

• 写入策略

  • 增量批处理：按天/小时批量写入，但归档到月分区。
  • 小文件控制：按 dt, region 分桶/聚合写入（coalesce/repartition），目标每分区 N 个文件（例如 1~10 个，依数据量而定）。
  • 去重策略（如需要）：以 click_id 为主键，在同分区内做去重（窗口函数或 merge-on-read 方案）。

三、生命周期（365 天）清理策略 问题：按月分区但生命周期按天计。需定义精确而可执行的删除规则，避免误删或超保留。

• 定义
  • 当前日期 D = current_date()
  • 阈值日 T = D - 365 天
  • 应删除的分区：其“整月”完全早于 T 的月份；等价于“删除所有 dt <= T 所在月份的上一个月份”
• 等价公式
  • droppable_dt_max = date_format(add_months(trunc(T, 'MM'), -1), 'yyyyMM')
  • 需删除的 dt 满足 dt <= droppable_dt_max
  • 直观解释：只要某个月的最后一天 < T，就可以整月删除。这样在边界月不出现“按天裁剪”的问题。
• 示例
  • 若今天为 2025-09-26，则 T=2024-09-26
  • trunc(T,'MM')=2024-09-01，上一月为 2024-08
  • 结论：删除 dt <= 202408 的所有分区（所有 region）

四、生命周期清理作业示例 A. Spark SQL 生成 dt 列表并批量删除（推荐做法，避免全表扫描）

1. 计算 droppable_dt_max with t as (select date_sub(current_date(), 365) as T) select date_format(add_months(trunc(T, 'MM'), -1), 'yyyyMM') as droppable_dt_max from t;

2. 获取待删 dt 列表（仅元数据扫描，不读取数据文件） select dt from (select distinct dt from dwd.dwd_clicks) p where dt <= '${droppable_dt_max}';

3. 对每个 dt 执行删除 -- Hive 支持对多级分区使用部分规范删除（按 dt 级联删除所有 region） alter table dwd.dwd_clicks drop if exists partition (dt='YYYYMM');

注：

• 如遇到 Hive 版本限制需指定全部分区列，可先枚举该月下所有 region，然后按 dt+region 组合删除： show partitions dwd.dwd_clicks partition (dt='YYYYMM'); -- 返回形如 dt=YYYYMM/region=us -- 解析出 region 值，逐一： alter table dwd.dwd_clicks drop if exists partition (dt='YYYYMM', region='us');

B. Hive CLI 简化方案（通过 shell 组装）

• 计算 droppable_dt_max，遍历 show partitions 或 select distinct dt 生成删除语句，批量执行。

五、自动化与频率

• 清理任务建议每日运行一次，避免堆积过期分区。
• 删除后不需要 MSCK REPAIR（Drop 会同步元数据）；如存在外部目录遗留，请确保垃圾回收或回收站配置正确（如 S3 生命周期或 HDFS Trash）。

六、边界与一致性处理

• 时区统一：派生 dt 前确保 event_time 已转为统一时区（建议 UTC）。
• 补到达/晚到数据
  • 晚到数据仍会进入历史月分区。TTL 清理按规则会在月粒度删除，不会出现“只删晚到数据”的情况。
  • 若晚到窗口超过 365 天，数据将写入已计划删除的分区；需在清理前完成该分区的最终写入，或引入业务例外（例如关闭超时晚到）。
• region 值规范
  • 建议对 region 做字典映射与校验；非法值统一落 unknown，避免产生稀疏/污染分区。
• 小文件/合并
  • 对热点月份与区域定期做分区级 compaction（按 dt, region 做 insert overwrite / repartition）。

七、查询与分区裁剪示例

• 强制下游查询带上 dt 或范围条件，最大化分区裁剪效果。 示例： select count(*) from dwd.dwd_clicks where dt between '202501' and '202507' and region in ('cn','us');

通过以上设计：

• 分区布局：dt（月）+ region（二级）
• 写入：动态分区、规范化 region、按月归档
• 生命周期：基于 T = 当前日期 - 365 天，删除所有 dt <= 上一个 T 月的月份
• 运行维护：每日清理，支持版本差异的 drop 粒度，确保元数据与文件一致性

以下为基于 bi_sales 的周分区 dt 与二级分区 biz_line 的数据分区逻辑设计与实现方案。目标是确保分区可准确映射业务周、便于分区裁剪与增量加载，并兼容 Hive/Spark/Trino 等主流引擎。

一、分区字段与命名规范

• 分区列：
  • dt：周分区，取该记录所属周的周一日期，格式为 yyyymmdd（例如 20250922）。
  • biz_line：业务线二级分区，字符串类型。
• 时间基准：
  • 使用 Asia/Shanghai（UTC+8）时区的业务事件时间（如 sales_ts 或订单支付时间 paid_ts）作为分区时间来源。
• 周定义：
  • ISO 周，周一 00:00 至下周一 00:00。dt 即该周的周一日期。

二、周分区计算规则

• 给定事件时间列 event_ts（timestamp），计算所属周周一：
  • Hive/Spark SQL（稳健写法，避免不同引擎对 week 起始日差异）： dt = date_format(date_sub(next_day(event_ts, 'MO'), 7), 'yyyyMMdd')
  • Trino/Presto SQL： dt = format_datetime(date_trunc('week', event_ts), 'yyyyMMdd')
• biz_line 来源：
  • 若 bi_sales 中已有 biz_line 字段，直接使用（需标准化）。
  • 若需从维表映射（如 dim_product/product_id -> biz_line），以左连接补全： COALESCE(dim.biz_line, raw.biz_line, 'unknown')。

三、字段标准化与容错

• biz_line 标准化：
  • 去首尾空格、转小写、空格转下划线、限定字符集（字母数字和下划线），为空则使用 'unknown'。
• 时间字段容错：
  • 对缺失或异常时间（event_ts IS NULL）的记录，直接落入修复队列或落在最近一周的 dt（不推荐）。建议丢弃或单独入未分区临时表进行人工修复。
• 示例（Spark/Hive）： biz_line_std = regexp_replace(lower(trim(biz_line)), '[^a-z0-9_]', '_') biz_line_final = nvl(nullif(biz_line_std, ''), 'unknown')

四、分区表 DDL 示例（Hive/Spark）

• 若 bi_sales 为明细事实表： CREATE TABLE bi_sales ( order_id string, event_ts timestamp, product_id string, amount decimal(18,2), ...其他业务字段... ) PARTITIONED BY (dt string, biz_line string) STORED AS PARQUET TBLPROPERTIES ('parquet.compression'='snappy');

• 注意：

  • 若需支持更新/去重（主键 upsert、迟到数据修正），建议使用 Iceberg/Hudi 并保留相同分区列： PARTITIONED BY (dt, biz_line) 并以 order_id 作为主键。

五、动态分区写入（增量装载）

• Hive/Spark SQL： set hive.exec.dynamic.partition=true; set hive.exec.dynamic.partition.mode=nonstrict;

  INSERT OVERWRITE TABLE bi_sales PARTITION (dt, biz_line) SELECT s.order_id, s.event_ts, s.product_id, s.amount, date_format(date_sub(next_day(s.event_ts, 'MO'), 7), 'yyyyMMdd') as dt, nvl(nullif(regexp_replace(lower(trim(coalesce(d.biz_line, s.biz_line))), '[^a-z0-9_]', '_'), ''), 'unknown') as biz_line FROM src_sales s LEFT JOIN dim_product d ON s.product_id = d.product_id WHERE s.event_ts IS NOT NULL;

• Spark DataFrame（等价逻辑）： df = spark.table("src_sales")
  .join(spark.table("dim_product"), "product_id", "left")
  .withColumn("dt", date_format(date_sub(next_day(col("event_ts"), "MO"), 7), "yyyyMMdd"))
  .withColumn("biz_line", expr("nvl(nullif(regexp_replace(lower(trim(coalesce(dim_product.biz_line, src_sales.biz_line))), '[^a-z0-9_]', '_'), ''), 'unknown')")) df.write.mode("overwrite").format("parquet").insertInto("bi_sales")

• Trino/Presto： INSERT INTO bi_sales SELECT order_id, event_ts, product_id, amount, format_datetime(date_trunc('week', event_ts), 'yyyyMMdd') AS dt, coalesce(nullif(regexp_replace(lower(trim(coalesce(d.biz_line, s.biz_line))), '[^a-z0-9_]', '_'), ''), 'unknown') AS biz_line FROM src_sales s LEFT JOIN dim_product d ON s.product_id = d.product_id WHERE s.event_ts IS NOT NULL;

六、迟到/修正数据处理策略

• 周期回补窗口：
  • 每日任务对最近 N 周（建议 2–4 周）的 dt 分区执行 INSERT OVERWRITE 或 upsert，以吸收迟到数据。
• 使用湖格式（推荐）：
  • Iceberg/Hudi 支持 MERGE/UPSERT 与时间旅行。示例（Spark + Hudi）： MERGE INTO bi_sales t USING new_data s ON t.order_id = s.order_id WHEN MATCHED THEN UPDATE SET ... WHEN NOT MATCHED THEN INSERT (..., dt, biz_line);
• 分区修复：
  • 修复时仅针对受影响的 dt 和 biz_line 分区，避免全表重算。

七、查询与分区裁剪示例

• 按周范围与业务线检索： SELECT ... FROM bi_sales WHERE dt BETWEEN '20250915' AND '20250929' AND biz_line IN ('retail', 'online');

• 周定位示例：

  • 2025-09-26（周五）落在 dt=20250922（周一）分区。

八、分区管理与监控

• 产出校验：
  • 每次装载后统计分区行数与金额汇总，与源系统对账。
• 分区可见性：
  • show partitions bi_sales where dt='20250922';
• 元数据兼容：
  • 保持 dt 为字符串类型，避免因日期类型在不同引擎的时区/序列化差异造成分区不可见。

九、注意事项

• 统一时区：作业和 SQL 会话需统一为 Asia/Shanghai。
• 数据倾斜：
  • 若 biz_line 取值过多导致小文件大量存在，考虑在存储层（如 Hudi/Iceberg）开启小文件合并或使用分桶辅助。
• 命名一致性：
  • dt 统一 yyyymmdd；biz_line 统一小写+下划线。

以上逻辑可直接用于构建 bi_sales 的周分区 dt 与 biz_line 二级分区的 ETL/查询方案，确保分区裁剪有效、迟到数据可回补、查询性能稳定。