Oracle 数据仓库 ETL 详解

适用版本：Oracle Database 10g / 11g / 12c / 19c / 23ai 文档版本：v1.0 / 2026-07

---

1. 概述

数据仓库 ETL（Extract-Transform-Load）[1]：

详细见：Oracle 数据仓库 ETL。

---

2. 架构

2.1 流程

源系统 → Extract → Staging → Transform → Load → 数据仓库

2.2 组件

- ODS（操作数据存储）
- Staging（暂存）
- Data Warehouse（仓库）
- Data Mart（数据集市）

---

3. Extract

3.1 全量

INSERT INTO stg_emp
SELECT * FROM source_emp@source_db;

3.2 增量

-- 基于时间
INSERT INTO stg_emp
SELECT * FROM source_emp@source_db
WHERE updated_at > :last_extract;

-- 基于 CDC
-- LogMiner / GoldenGate

3.3 外部表

CREATE TABLE ext_sales (
  id NUMBER,
  amount NUMBER,
  sale_date DATE
)
ORGANIZATION EXTERNAL (
  TYPE ORACLE_LOADER
  DEFAULT DIRECTORY data_dir
  ACCESS PARAMETERS (
    RECORDS DELIMITED BY NEWLINE
    FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"'
  )
  LOCATION ('sales_2025.csv')
);

SELECT * FROM ext_sales;

---

4. Transform

4.1 清洗

-- 去重
DELETE FROM stg_emp WHERE ROWID IN (
  SELECT rid FROM (
    SELECT ROWID rid, ROW_NUMBER() OVER (PARTITION BY email ORDER BY id) rn
    FROM stg_emp
  ) WHERE rn > 1
);

-- 标准化
UPDATE stg_emp SET name = TRIM(UPPER(name));
UPDATE stg_emp SET phone = REGEXP_REPLACE(phone, '[^0-9]', '');

-- 默认值
UPDATE stg_emp SET status = NVL(status, 'ACTIVE');

4.2 转换

-- 类型转换
UPDATE stg_emp SET salary = TO_NUMBER(salary_str);

-- 日期
UPDATE stg_emp SET hire_date = TO_DATE(hire_str, 'YYYY-MM-DD');

-- 拆分
INSERT INTO emp_addr (emp_id, address)
SELECT id, address FROM stg_emp WHERE address IS NOT NULL;

-- 合并
INSERT INTO emp_full (id, name, addr, phone)
SELECT e.id, e.name, a.address, p.phone
FROM stg_emp e
  LEFT JOIN emp_addr a ON e.id = a.emp_id
  LEFT JOIN emp_phone p ON e.id = p.emp_id;

4.3 查找

-- 维度查找
UPDATE fact_sales s
SET dept_id = (
  SELECT d.id FROM dim_dept d 
  WHERE d.name = s.dept_name
);

-- MERGE
MERGE INTO fact_sales f
USING dim_dept d
ON (f.dept_name = d.name)
WHEN MATCHED THEN UPDATE SET f.dept_id = d.id;

详细见：Oracle MERGE 语句详解。

4.4 聚合

INSERT INTO agg_sales_monthly (year, month, dept_id, total)
SELECT EXTRACT(YEAR FROM sale_date),
       EXTRACT(MONTH FROM sale_date),
       dept_id,
       SUM(amount)
FROM fact_sales
GROUP BY EXTRACT(YEAR FROM sale_date), EXTRACT(MONTH FROM sale_date), dept_id;

---

5. Load

5.1 直接路径

-- INSERT /*+ APPEND */
INSERT /*+ APPEND */ INTO sales
SELECT * FROM stg_sales;

-- SQL*Loader
-- DIRECT=TRUE

5.2 分区交换

-- 准备表
CREATE TABLE sales_2025_07 (...) AS SELECT * FROM sales WHERE 1=0;

-- 加载
INSERT /*+ APPEND */ INTO sales_2025_07 SELECT * FROM stg_2025_07;

-- 交换
ALTER TABLE sales EXCHANGE PARTITION p2025_07 WITH TABLE sales_2025_07;

5.3 批量

-- BULK
DECLARE
  TYPE sales_tab IS TABLE OF sales%ROWTYPE;
  v_sales sales_tab;
BEGIN
  SELECT * BULK COLLECT INTO v_sales FROM stg_sales LIMIT 10000;
  
  FORALL i IN 1..v_sales.COUNT
    INSERT INTO sales VALUES v_sales(i);
  
  COMMIT;
END;
/

详细见：Oracle BULK COLLECT 与 FORALL 详解。

---

6. Data Pump

6.1 Export

expdp scott/tiger DIRECTORY=dp_dir DUMPFILE=sales.dmp TABLES=sales
expdp scott/tiger DIRECTORY=dp_dir DUMPFILE=full.dmp FULL=Y
expdp scott/tiger DIRECTORY=dp_dir DUMPFILE=schema.dmp SCHEMAS=scott

6.2 Import

impdp scott/tiger DIRECTORY=dp_dir DUMPFILE=sales.dmp TABLES=sales
impdp scott/tiger DIRECTORY=dp_dir DUMPFILE=sales.dmp TABLE_EXISTS_ACTION=APPEND
impdp scott/tiger DIRECTORY=dp_dir DUMPFILE=full.dmp REMAP_SCHEMA=scott:hr

详细见：Oracle Data Pump 全集。

---

7. SQL*Loader

7.1 控制文件

LOAD DATA
INFILE 'sales.csv'
BADFILE 'sales.bad'
DISCARDFILE 'sales.dsc'
APPEND INTO TABLE sales
FIELDS TERMINATED BY ',' OPTIONALLY ENCLOSED BY '"'
TRAILING NULLCOLS
(
  id INTEGER,
  amount DECIMAL,
  sale_date DATE 'YYYY-MM-DD',
  status CONSTANT 'NEW'
)

7.2 命令

sqlldr scott/tiger control=sales.ctl log=sales.log
sqlldr scott/tiger control=sales.ctl direct=true  -- 直接路径

---

8. External Table

8.1 创建

CREATE TABLE ext_sales (
  id NUMBER,
  amount NUMBER,
  sale_date DATE
)
ORGANIZATION EXTERNAL (
  TYPE ORACLE_LOADER
  DEFAULT DIRECTORY data_dir
  ACCESS PARAMETERS (
    RECORDS DELIMITED BY NEWLINE
    FIELDS TERMINATED BY ','
    MISSING FIELD VALUES ARE NULL
  )
  LOCATION ('sales_2025.csv')
)
REJECT LIMIT UNLIMITED;

8.2 使用

-- 查询
SELECT * FROM ext_sales;

-- 加载
INSERT /*+ APPEND */ INTO sales SELECT * FROM ext_sales;

---

9. 物化视图

9.1 聚合

CREATE MATERIALIZED VIEW mv_sales_monthly
  REFRESH COMPLETE ON DEMAND
  ENABLE QUERY REWRITE
  AS
  SELECT EXTRACT(YEAR FROM sale_date) AS yr,
         EXTRACT(MONTH FROM sale_date) AS mon,
         dept_id,
         SUM(amount) AS total
  FROM sales
  GROUP BY EXTRACT(YEAR FROM sale_date), EXTRACT(MONTH FROM sale_date), dept_id;

9.2 查询重写

ALTER SESSION SET query_rewrite_enabled = TRUE;

-- 自动使用物化视图
SELECT dept_id, SUM(amount) FROM sales GROUP BY dept_id;

详细见：Oracle 视图与物化视图详解。

---

10. Dimension

CREATE DIMENSION time_dim
  LEVEL day IS time.day
  LEVEL month IS time.month
  LEVEL quarter IS time.quarter
  LEVEL year IS time.year
  HIERARCHY time_rollup (
    day CHILD OF month CHILD OF quarter CHILD OF year
  )
  ATTRIBUTE month DETERMINES month_name;

---

11. Partition

11.1 Range

CREATE TABLE sales (
  id NUMBER,
  sale_date DATE,
  amount NUMBER
)
PARTITION BY RANGE (sale_date) (
  PARTITION p2024 VALUES LESS THAN (TO_DATE('2025-01-01', 'YYYY-MM-DD')),
  PARTITION p2025 VALUES LESS THAN (TO_DATE('2026-01-01', 'YYYY-MM-DD')),
  PARTITION p2026 VALUES LESS THAN (TO_DATE('2027-01-01', 'YYYY-MM-DD'))
);

11.2 交换

ALTER TABLE sales EXCHANGE PARTITION p2025 WITH TABLE sales_stg;

详细见：Oracle 表分区策略详解。

---

12. 压缩

12.1 压缩类型

-- 仓库压缩
CREATE TABLE sales (...) COMPRESS FOR QUERY HIGH;

-- 归档
CREATE TABLE sales_archive (...) COMPRESS FOR ARCHIVE HIGH;

详细见：Oracle 表压缩技术详解。

---

13. 调度

BEGIN
  DBMS_SCHEDULER.CREATE_JOB(
    job_name => 'etl_daily',
    job_type => 'PLSQL_BLOCK',
    job_action => 'BEGIN etl_pkg.run_daily; END;',
    start_date => SYSTIMESTAMP,
    repeat_interval => 'FREQ=DAILY; BYHOUR=2',
    enabled => TRUE
  );
END;
/

---

14. 错误处理

14.1 日志表

CREATE TABLE etl_log (
  id NUMBER GENERATED ALWAYS AS IDENTITY,
  job_name VARCHAR2(100),
  step VARCHAR2(100),
  status VARCHAR2(20),
  error_msg VARCHAR2(4000),
  rows_processed NUMBER,
  start_time TIMESTAMP,
  end_time TIMESTAMP
);

14.2 SAVE EXCEPTIONS

FORALL i IN 1..v_data.COUNT SAVE EXCEPTIONS
  INSERT INTO target VALUES v_data(i);

EXCEPTION
  WHEN OTHERS THEN
    FOR i IN 1..SQL%BULK_EXCEPTIONS.COUNT LOOP
      INSERT INTO etl_log (...) VALUES (...);
    END LOOP;

详细见：Oracle BULK COLLECT 与 FORALL 详解。

---

15. 性能

15.1 并行

ALTER SESSION ENABLE PARALLEL DML;
INSERT /*+ PARALLEL(s 8) APPEND */ INTO sales s SELECT * FROM stg_sales;

15.2 直接路径

INSERT /*+ APPEND */ INTO sales SELECT * FROM stg_sales;

15.3 分区

- 分区交换
- 局部索引
- 增量

详细见：Oracle 并行查询详解。

---

16. 监控

16.1 进度

SELECT sid, serial#, opname, sofar, totalwork, ROUND(sofar/totalwork*100, 2) AS pct
FROM v$session_longops
WHERE opname LIKE 'ETL%';

16.2 统计

SELECT * FROM etl_log WHERE job_name = 'etl_daily' ORDER BY start_time DESC;

---

17. 常见坑与排错

17.1 数据倾斜

- 并行不均
- 分区
- Skew

17.2 约束违反

- 检查数据
- 异常表
- 清洗

17.3 性能慢

- 索引过多
- 触发器
- 直接路径

---

18. 最佳实践

1. 外部表：灵活
2. 直接路径：性能
3. 分区交换：高效
4. 物化视图：聚合
5. 并行：吞吐
6. 压缩：空间
7. MERGE：UPSERT
8. BULK：批量
9. 错误处理：完整
10. 监控：进度

---

19. 参考资料

[1] Oracle Database Data Warehousing Guide 19c https://docs.oracle.com/en/database/oracle/oracle-database/19/dwh/