数据分区逻辑设计

以下为基于 ods_orders 的数据分区与分桶设计及实现逻辑，满足 dt（日分区）与按 user_id 16 桶的要求。

一、分区与分桶规则
- 分区键 dt：
  - 来源优先使用 ods_orders 的分区字段 dt（字符串，格式 yyyy-MM-dd）。
  - 若 ods_orders 未提供 dt 字段或格式不一致，则以订单事件时间（如 order_create_time）转换为日期：dt = date_format(order_create_time, 'yyyy-MM-dd')。
- 分桶键 user_id：
  - 按 user_id 进行 16 桶分桶。
  - 同一 user_id 的记录通过哈希取模分配到同一桶，保证同用户数据在单桶内，有利于后续按 user_id 的 Join/聚合。
- 异常与边界处理：
  - user_id 为空时建议置为保留值（如 -1）防止空值分布不确定：user_id_norm = coalesce(user_id, -1)。
  - dt 非法或缺失记录应过滤或重定向到异常分区（如 dt='0000-00-00'，不推荐），优先过滤。
- 文件格式与压缩建议：
  - 使用 ORC 或 Parquet（推荐 ORC + SNAPPY 压缩），利于分区裁剪与列式存储优化。

二、Hive 表结构与建表语句（示例）
- 目标层假设为 dwd_orders（可根据实际层级命名：ods/dwd/dws）。
- 以 ORC、分区 dt、按 user_id 分桶 16。

CREATE TABLE IF NOT EXISTS dwd_orders (
  order_id         string,
  user_id          bigint,
  item_id          string,
  order_status     string,
  amount           decimal(20,4),
  order_create_time timestamp
)
PARTITIONED BY (dt string)
CLUSTERED BY (user_id) INTO 16 BUCKETS
STORED AS ORC
TBLPROPERTIES (
  'orc.compress'='SNAPPY'
);

三、数据装载逻辑（动态分区 + 强制分桶）
- 加载同日/同范围分区数据时，使用动态分区与强制分桶设置。

-- 开启动态分区与分桶
SET hive.exec.dynamic.partition=true;
SET hive.exec.dynamic.partition.mode=nonstrict;
SET hive.enforce.bucketing=true;

-- 可选：限制动态分区数量
SET hive.exec.max.dynamic.partitions=10000;
SET hive.exec.max.dynamic.partitions.pernode=1000;

-- 从 ODS 按 dt 装载（优先使用 ODS 的 dt 分区）
INSERT OVERWRITE TABLE dwd_orders PARTITION (dt)
SELECT
  o.order_id,
  COALESCE(o.user_id, -1) AS user_id,      -- 异常值处理
  o.item_id,
  o.order_status,
  o.amount,
  o.order_create_time,
  o.dt                                       -- 直接沿用 ODS 分区字段
FROM ods_orders o
WHERE o.dt BETWEEN '2025-09-01' AND '2025-09-30';

-- 若 ODS 不含 dt 字段，改用事件时间生成 dt
INSERT OVERWRITE TABLE dwd_orders PARTITION (dt)
SELECT
  o.order_id,
  COALESCE(o.user_id, -1) AS user_id,
  o.item_id,
  o.order_status,
  o.amount,
  o.order_create_time,
  date_format(o.order_create_time, 'yyyy-MM-dd') AS dt
FROM ods_orders o
WHERE o.order_create_time >= '2025-09-01' AND o.order_create_time < '2025-10-01';

四、分区与分桶行为说明
- 分区裁剪：查询时加 dt 过滤条件（WHERE dt='2025-09-25' 或范围）可显著减少扫描。
- 分桶分配：Hive 对 user_id 进行哈希后对 16 取模分配桶，确保同一 user_id 落入稳定桶。开启 hive.enforce.bucketing 可保证写入时生成 16 个桶文件。
- 目录结构示例：
  - .../dwd_orders/dt=2025-09-25/000000_0（bucket-0）
  - .../dwd_orders/dt=2025-09-25/000001_0（bucket-1）
  - ...
  - .../dwd_orders/dt=2025-09-25/000015_0（bucket-15）

五、增量与维护建议
- 每日增量装载：按当日日分区 dt 进行 INSERT OVERWRITE/INSERT INTO（根据是否需要重算）。
- 异常数据治理：
  - 过滤 dt 为空或非法记录：WHERE dt IS NOT NULL AND dt RLIKE '^[0-9]{4}-[0-9]{2}-[0-9]{2}$'。
  - user_id 为空统一映射为 -1，后续可在指标口径中排除或标记。
- 分区修复与清理：
  - 外部新增分区后可执行 MSCK REPAIR TABLE dwd_orders 修复元数据。
  - 定期删除过期分区：ALTER TABLE dwd_orders DROP PARTITION (dt='2025-06-01');（逐日执行或脚本化）。

六、查询优化要点
- 任何查询都应带 dt 条件以触发分区裁剪。
- 以 user_id 为 Join 键时，与同样分桶策略的表进行 Bucket Map Join 可降低 Shuffle 成本（需两表相同桶数与分桶键）。

上述设计确保以 dt 进行日级分区并按 user_id 进行 16 桶分桶，兼顾加载与查询性能，同时提供清晰可维护的装载与治理流程。

以下为 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/查询方案，确保分区裁剪有效、迟到数据可回补、查询性能稳定。