# 大宽表设计

> 提能任务项目 - Spark SQL 大宽表设计方案

---

## 📋 设计概述

### 设计目标

将提能任务相关的 4 个表（task、sdc_feedback、verify_feedback、bpm_record）整合为一张大宽表，便于：
- ✅ 数据分析与报表
- ✅ 机器学习特征工程
- ✅ 即席查询
- ✅ BI 工具直连

### 设计原则

| 原则 | 说明 |
|------|------|
| **Header 优先** | 以 `enhance_task` 为主表 |
| **一对多展开** | sdc_feedback 和 verify_feedback 多条记录展开为多行 |
| **最新审批** | bpm_record 只取最新一条 |
| **字段前缀** | 表前缀_原字段名（t_, s_, v_, b_） |

---

## 🗂️ 表结构定义

### 大宽表字段清单

```sql
CREATE TABLE IF NOT EXISTS dwd_capacity_enhance_wide (
    -- ========== 任务主表字段 (t_) ==========
    t_id                          BIGINT       COMMENT '主键 ID',
    t_task_id                     STRING       COMMENT '业务键 - 任务 ID',
    t_task_name                   STRING       COMMENT '任务名称',
    t_task_status                 TINYINT      COMMENT '任务状态 (0-11)',
    t_task_status_name            STRING       COMMENT '任务状态名称',
    t_supplier_code               STRING       COMMENT '供应商代码',
    t_factory_code                STRING       COMMENT '工厂代码',
    t_material_code               STRING       COMMENT '物料代码',
    t_supplier_responsible        STRING       COMMENT '供应商负责人',
    t_operation_responsible       STRING       COMMENT '运营负责人',
    t_quality_responsible         STRING       COMMENT '质量负责人',
    t_purchase_responsible        STRING       COMMENT '采购负责人',
    t_sdc_deadline_date           DATE         COMMENT 'SDC 截止日期',
    t_verify_deadline_date        DATE         COMMENT '验证截止日期',
    t_created_at                  TIMESTAMP    COMMENT '创建时间',
    t_updated_at                  TIMESTAMP    COMMENT '更新时间',
    
    -- ========== SDC 反馈表字段 (s_) ==========
    s_id                          BIGINT       COMMENT 'SDC 反馈主键',
    s_promised_sdc                DECIMAL(18,4) COMMENT '承诺产能 (SDC)',
    s_enhance_cost                DECIMAL(18,2) COMMENT '提能代价/成本',
    s_reach_date                  DATE         COMMENT '达产日期',
    s_deviation_reason            STRING       COMMENT '偏差原因',
    s_enhance_plan                STRING       COMMENT '提能方案',
    s_created_at                  TIMESTAMP    COMMENT 'SDC 反馈创建时间',
    
    -- ========== 验证产能反馈表字段 (v_) ==========
    v_id                          BIGINT       COMMENT '验证反馈主键',
    v_verify_capacity             DECIMAL(18,4) COMMENT '验证产能 (实际值)',
    v_enhance_cost                DECIMAL(18,2) COMMENT '实际提能成本',
    v_reach_date                  DATE         COMMENT '实际达产日期',
    v_deviation_rate              DECIMAL(8,4)  COMMENT '偏差率 (%)',
    v_deviation_level             STRING       COMMENT '偏差等级',
    v_created_at                  TIMESTAMP    COMMENT '验证反馈创建时间',
    
    -- ========== BPM 审批表字段 (b_) ==========
    b_id                          BIGINT       COMMENT '审批记录主键',
    b_approval_type               TINYINT      COMMENT '审批类型 (1/2/3)',
    b_approval_type_name          STRING       COMMENT '审批类型名称',
    b_approval_round              INT          COMMENT '审批轮次',
    b_approval_result             TINYINT      COMMENT '审批结果 (0/1/2/3)',
    b_approval_result_name        STRING       COMMENT '审批结果名称',
    b_model_type                  STRING       COMMENT '模型类型',
    b_model_code                  STRING       COMMENT '模型代码',
    b_process_instance_status     STRING       COMMENT '流程实例状态',
    b_bpm_business_key            STRING       COMMENT 'BPM 业务键',
    b_created_at                  TIMESTAMP    COMMENT '审批创建时间',
    
    -- ========== 衍生字段 ==========
    dw_created_date               DATE         COMMENT '数据仓库分区日期',
    dw_version                    STRING       COMMENT '数据版本'
)
COMMENT '提能任务大宽表'
PARTITIONED BY (dw_created_date DATE)
STORED AS ORC
TBLPROPERTIES ('orc.compress' = 'SNAPPY');
```

---

## 🔧 ETL 逻辑

### Spark SQL 实现

```sql
-- ========== 提能任务大宽表 ETL ==========
INSERT OVERWRITE TABLE dwd_capacity_enhance_wide 
PARTITION (dw_created_date = '${biz_date}')
SELECT 
    -- 任务主表字段
    t.id                              AS t_id,
    t.task_id                         AS t_task_id,
    t.task_name                       AS t_task_name,
    t.task_status                     AS t_task_status,
    CASE t.task_status
        WHEN 0 THEN '待发布'
        WHEN 1 THEN '调研中'
        WHEN 2 THEN '任务取消'
        WHEN 3 THEN '待审批 SDC'
        WHEN 4 THEN 'SDC 驳回'
        WHEN 5 THEN '待反馈验证产能'
        WHEN 6 THEN '待审批验证产能'
        WHEN 7 THEN '验证产能驳回'
        WHEN 8 THEN '验证产能有偏差'
        WHEN 9 THEN '无效提能'
        WHEN 10 THEN '偏差关闭'
        WHEN 11 THEN '提能完成'
        ELSE '未知'
    END                               AS t_task_status_name,
    t.supplier_code                   AS t_supplier_code,
    t.factory_code                    AS t_factory_code,
    t.material_code                   AS t_material_code,
    t.supplier_responsible            AS t_supplier_responsible,
    t.operation_responsible           AS t_operation_responsible,
    t.quality_responsible             AS t_quality_responsible,
    t.purchase_responsible            AS t_purchase_responsible,
    t.sdc_deadline_date               AS t_sdc_deadline_date,
    t.verify_deadline_date            AS t_verify_deadline_date,
    t.created_at                      AS t_created_at,
    t.updated_at                      AS t_updated_at,
    
    -- SDC 反馈表字段
    s.id                              AS s_id,
    s.promised_sdc                    AS s_promised_sdc,
    s.enhance_cost                    AS s_enhance_cost,
    s.reach_date                      AS s_reach_date,
    s.deviation_reason                AS s_deviation_reason,
    s.enhance_plan                    AS s_enhance_plan,
    s.created_at                      AS s_created_at,
    
    -- 验证产能反馈表字段
    v.id                              AS v_id,
    v.verify_capacity                 AS v_verify_capacity,
    v.enhance_cost                    AS v_enhance_cost,
    v.reach_date                      AS v_reach_date,
    -- 偏差率计算
    CASE 
        WHEN s.promised_sdc > 0 
        THEN ROUND((v.verify_capacity - s.promised_sdc) / s.promised_sdc * 100, 4)
        ELSE NULL 
    END                               AS v_deviation_rate,
    -- 偏差等级
    CASE 
        WHEN v.verify_capacity >= s.promised_sdc THEN '达标'
        WHEN v.verify_capacity >= s.promised_sdc * 0.9 THEN '轻微偏差'
        WHEN s.promised_sdc > 0 THEN '严重偏差'
        ELSE NULL
    END                               AS v_deviation_level,
    v.created_at                      AS v_created_at,
    
    -- BPM 审批表字段（最新一条）
    b.id                              AS b_id,
    b.approval_type                   AS b_approval_type,
    CASE b.approval_type
        WHEN 1 THEN 'SDC 无偏差采购审批'
        WHEN 2 THEN 'SDC 有偏差三方会签'
        WHEN 3 THEN '验证产能 SQE 审批'
        ELSE '未知'
    END                               AS b_approval_type_name,
    b.approval_round                  AS b_approval_round,
    b.approval_result                 AS b_approval_result,
    CASE b.approval_result
        WHEN 0 THEN '处理中'
        WHEN 1 THEN '通过'
        WHEN 2 THEN '驳回'
        WHEN 3 THEN '终止/撤回'
        ELSE '未知'
    END                               AS b_approval_result_name,
    b.model_type                      AS b_model_type,
    b.model_code                      AS b_model_code,
    b.process_instance_status         AS b_process_instance_status,
    b.bpm_business_key                AS b_bpm_business_key,
    b.created_at                      AS b_created_at,
    
    -- 衍生字段
    TO_DATE(t.created_at)             AS dw_created_date,
    '${version}'                      AS dw_version

FROM m_capacity_enhance_task t

-- 关联 SDC 反馈表（1:N 展开）
LEFT JOIN m_capacity_enhance_sdc_feedback s 
    ON t.task_id = s.enhance_task_id

-- 关联验证产能反馈表（1:N 展开）
LEFT JOIN m_capacity_enhance_verify_feedback v 
    ON s.id = v.sdc_feedback_id

-- 关联最新 BPM 审批记录
LEFT JOIN (
    SELECT 
        enhance_task_id,
        id,
        approval_type,
        approval_round,
        approval_result,
        model_type,
        model_code,
        process_instance_status,
        bpm_business_key,
        created_at,
        ROW_NUMBER() OVER (
            PARTITION BY enhance_task_id 
            ORDER BY created_at DESC
        ) AS rn
    FROM m_capacity_enhance_bpm_record
    WHERE created_at >= DATE_SUB(CURRENT_DATE, 365)  -- 只取近一年数据
) b ON t.task_id = b.enhance_task_id AND b.rn = 1

WHERE t.created_at >= DATE_SUB(CURRENT_DATE, 365)  -- 增量同步
  AND t.created_at < DATE_ADD(CURRENT_DATE, 1);
```

---

## 📊 数据关系示意

### 一对多展开逻辑

```
原始数据:
┌─────────────┐
│ Task-001    │
├─────────────┤
│ SDC-001     │──┐
│ SDC-002     │──┼── 1 个任务有 2 条 SDC 反馈
└─────────────┘  │
                 │
┌─────────────┐  │
│ SDC-001     │──┘
├─────────────┤
│ Verify-001  │──┐
│ Verify-002  │──┼── 1 条 SDC 有 2 条验证反馈
└─────────────┘  │
                 │
┌─────────────┐  │
│ SDC-002     │──┘
├─────────────┤
│ Verify-003  │─── 1 条 SDC 有 1 条验证反馈
└─────────────┘

展开后大宽表 (5 行):
┌─────────────┬───────────┬────────────┬───────────┐
│ Task-001    │ SDC-001   │ Verify-001 │ BPM-Latest│
│ Task-001    │ SDC-001   │ Verify-002 │ BPM-Latest│
│ Task-001    │ SDC-002   │ Verify-003 │ BPM-Latest│
└─────────────┴───────────┴────────────┴───────────┘
```

---

## 🔍 常用查询示例

### 1. 查询任务完整信息

```sql
SELECT 
    t_task_id,
    t_task_name,
    t_task_status_name,
    t_supplier_code,
    s_promised_sdc,
    v_verify_capacity,
    v_deviation_rate,
    v_deviation_level,
    b_approval_type_name,
    b_approval_result_name
FROM dwd_capacity_enhance_wide
WHERE dw_created_date = '2026-03-11'
  AND t_task_id = 'TASK-2026-001';
```

### 2. 偏差分析统计

```sql
SELECT 
    v_deviation_level,
    COUNT(DISTINCT t_task_id) AS task_count,
    AVG(v_deviation_rate) AS avg_deviation_rate,
    AVG(s_promised_sdc) AS avg_promised_sdc,
    AVG(v_verify_capacity) AS avg_verify_capacity
FROM dwd_capacity_enhance_wide
WHERE dw_created_date = '2026-03-11'
  AND v_deviation_level IS NOT NULL
GROUP BY v_deviation_level;
```

### 3. 审批通过率分析

```sql
SELECT 
    b_approval_type_name,
    COUNT(DISTINCT t_task_id) AS total_tasks,
    SUM(CASE WHEN b_approval_result = 1 THEN 1 ELSE 0 END) AS pass_count,
    ROUND(
        SUM(CASE WHEN b_approval_result = 1 THEN 1 ELSE 0 END) * 100.0 / COUNT(DISTINCT t_task_id),
        2
    ) AS pass_rate
FROM dwd_capacity_enhance_wide
WHERE dw_created_date = '2026-03-11'
  AND b_approval_type IS NOT NULL
GROUP BY b_approval_type_name;
```

### 4. 供应商绩效分析

```sql
SELECT 
    t_supplier_code,
    COUNT(DISTINCT t_task_id) AS task_count,
    SUM(CASE WHEN t_task_status = 11 THEN 1 ELSE 0 END) AS completed_count,
    ROUND(
        SUM(CASE WHEN t_task_status = 11 THEN 1 ELSE 0 END) * 100.0 / COUNT(DISTINCT t_task_id),
        2
    ) AS completion_rate,
    AVG(v_deviation_rate) AS avg_deviation_rate
FROM dwd_capacity_enhance_wide
WHERE dw_created_date >= '2026-01-01'
GROUP BY t_supplier_code
ORDER BY task_count DESC
LIMIT 20;
```

---

## ⚙️ 调度配置

### Airflow DAG 示例

```python
from airflow import DAG
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2026, 3, 11),
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'dwd_capacity_enhance_wide_etl',
    default_args=default_args,
    description='提能任务大宽表 ETL',
    schedule_interval='0 2 * * *',  # 每天凌晨 2 点
    catchup=False,
)

spark_job = SparkSubmitOperator(
    task_id='spark_capacity_enhance_wide',
    application='/opt/spark/jobs/capacity_enhance_wide_etl.py',
    conf={
        'spark.sql.shuffle.partitions': '200',
        'spark.executor.memory': '4g',
        'spark.executor.cores': '2',
    },
    application_args=['--biz-date', '{{ ds }}'],
    dag=dag,
)
```

---

## 📝 注意事项

### 数据质量

1. **去重检查**：确保大宽表无重复记录
2. **空值处理**：LEFT JOIN 可能产生 NULL，需妥善处理
3. **数据一致性**：源表变更后需重新计算衍生字段

### 性能优化

1. **分区管理**：按 dw_created_date 分区，定期清理旧数据
2. **文件压缩**：使用 ORC + SNAPPY 压缩
3. **小文件合并**：定期合并小文件

### 字段扩展

1. **新增字段**：在表末尾添加，保持向后兼容
2. **字段废弃**：标记 deprecated，不直接删除
3. **版本管理**：dw_version 字段记录数据版本

---

_最后更新：2026-03-11_
