Oracle 技术网 (OTN) > 面向开发人员和数据库管理员的下载、讨论和文档

 
下载 文档 论坛 文章 示例代码 培训 RSS 资源
使用 Oracle 数据库 10g 内部的 ETL 基础架构

OBE 主页 > 10gR2 VMware > 商务智能 > 使用 Oracle 数据库 10g 内部的 ETL 基础架构

使用 Oracle 数据库 10g 内部的 ETL 基础架构

目的

本教程介绍了 Oracle 数据库 10g 的提取、转换和加载 (ETL) 基础架构。

所需时间

大约 1 个小时

主题

本教程包括下列主题：

概述
为销售历史模式实施模式更改
了解多表插入
了解 DML 错误记录功能
了解表函数的基础知识
使用更改数据捕获 (CDC) 跟踪和使用增量式数据更改
为数据集市填充信息
清理
总结

查看屏幕截图

将光标置于此图标上以加载和查看本教程的所有屏幕截图。（警告：因为此操作会同时加载所有屏幕截图，所以网速较慢时，响应时间可能会比较长。）

注意：此外，您还可以在下列步骤中将光标放在每个单独的图标上，从而仅加载和查看与该步骤相关的屏幕截图。您可以单击单个屏幕截图将其隐藏。

概述

ETL 过程中执行哪些操作？

ETL 表示提取、转换和加载。在提取过程中，必须从许多不同的数据源（包括数据库系统和应用程序）中标识并提取所需的数据。通常情况下，无法标识相关的特定数据子集，这意味着必须先提取多余的数据，并在稍后的时刻标识相关数据。根据源系统的功能（例如，OS 资源），此提取过程中可能会发生某些转换。提取的数据大小少则数百 KB，多则数百 GB，具体情况取决于源系统和业务情况。两个（逻辑上）相同的提取之间的时间间隔也存在这种情况：时间跨度从数天/小时和分钟一直到近乎实时。例如，Web 服务器日志文件的大小可在极短的时间内轻松地增至数百 MB。

提取数据后，需要将其物理传输到目标系统或中间系统进行进一步的处理。根据所选择的传输方法，在此过程中也可能会进行某些转换。例如，通过网关直接访问远程目标的 SQL 语句可以在 SELECT 语句中连接两列。

提取并传输数据后，将执行 ETL 中最具挑战性（并且最耗时）的过程：转换并加载到目标系统中。该过程可能包括：

应用复杂的过滤器

必须对照目标数据库表中已经存在的信息验证数据
必须对照目标对象检查在不知道新信息与已更改信息的情况下提取的数据，以决定是否必须更新或插入该数据
必须以细节级别和聚合信息的形式多次插入同一数据

该过程应在可伸缩的方式下尽快完成，并且不能影响为检索信息而对现有目标进行的并发访问。

Oracle 提供了各种功能来解决 ETL 情形中所有相关的问题和任务。Oracle 数据库 10g 是 ETL 转换引擎。

返回主题列表

为销售历史模式实施模式更改

在开始执行本 OBE 的任务之前，需要对现有的销售历史模式实施一些更改。还需要其他一些对象，并且必须将其他系统权限授予用户 SH。用于应用这些更改的 SQL 文件为 modifySH_10gR2.sql。您可以在 /home/oracle/wkdir 文件夹中找到该文件。要使用数据仓库教程的设置文件，执行下列步骤：

1.

启动终端窗口。从终端会话中执行以下命令，将工作目录更改为 /home/oracle/wkdir 文件夹：

注意：本教程假设您拥有一个 /home/oracle/wkdir 文件夹，其中包含可供使用的物化视图文件。

cd wkdir

 
2.

启动 SQL*Plus 会话。使用 sh/sh 作为用户名和口令。

从 SQL*Plus 会话中运行 modifySH_10gR2.sql 脚本。 

@modifySH_10gR2.sql

输出的结尾应与下图相符。

 

 

返回主题列表

了解多表插入

MyCompany 从它的某个合作伙伴公司那里收到一些非关系数据结构，该公司当时在销售某个特殊广告活动所宣传的产品。该数据结构是从专有大型机系统中提取的一个非规范化的非关系记录结构，并按照每个客户和每周的产品分别对应一个记录的方式构成。而且必须将这些数据结构插入到数据仓库中。由于销售记录数据是按照每个客户和每天的产品进行存储的，因此需要对传入的数据进行转换。

作为转换过程的一部分，必须将非关系的非规范化数据结构从每周一个记录转换为每周七个记录，每个记录包含一个工作日的信息（第一个业务转换示例）。此外，数据仓库还跟踪信贷限额超过某个限制的所有新客户。分别跟踪这些客户。

在本部分中，您将利用 Oracle 新的多表插入功能实施这些业务转换。为此，执行以下步骤：

使用多表插入实现旋转
使用多表插入实现条件插入

返回主题列表

使用多表插入实现旋转

1.

显示新多表插入的执行计划。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@explain_mti_new.sql

DELETE
FROM PLAN_TABLE;
EXPLAIN PLAN FOR
INSERT ALL
INTO sales
VALUES(product_id, customer_id,weekly_start_date,2,9999, q_sun,sales_sun)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+1,2,9999, q_mon,sales_mon)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+2,2,9999, q_tue,sales_tue)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+3,2,9999, q_wed,sales_wed)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+4,2,9999, q_thu,sales_thu)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+5,2,9999, q_fri,sales_fri)
INTO sales
VALUES(product_id, customer_id,weekly_start_date+6,2,9999, q_sat,sales_sat)
SELECT * FROM sales_input_table;
SET linesize 140
SELECT * from table(dbms_xplan.display);

注意：只对输入源表扫描一次！由于非规范化的复杂性在多个 INSERT INTO 分支中得到处理，从而避免多次扫描。

 

2.

现在，显示基于 UNION ALL 设置操作的多表插入的执行计划。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@explain_mti_old.sql

DELETE FROM PLAN_TABLE;
COMMIT;
EXPLAIN PLAN FOR
INSERT INTO sales
(prod_id, cust_id, time_id, channel_id,promo_id,amount_sold,quantity_sold)
SELECT product_id, customer_id,weekly_start_date,2,9999,sales_sun,q_sun
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+1,2,9999,sales_mon,q_mon
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+2,2,9999,sales_tue,q_tue
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+3,2,9999,sales_wed,q_wed
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+4,2,9999,sales_thu,q_thu
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+5,2,9999,sales_fri,q_fri
FROM sales_input_table
UNION ALL
SELECT product_id, customer_id,weekly_start_date+6,2,9999,sales_sat,q_sat
FROM sales_input_table;
SET linesize 140
SELECT * from table(dbms_xplan.display);
COMMIT;

注意：对输入源表扫描七次！非规范化的复杂性在多个 SELECT 操作中得到处理。

With an increasing number of input records, the superiority and the performance improvement of the new multi-table insert statement—by reducing the statement to only one SCAN—will become more and more obvious.

返回主题

使用多表插入实现条件插入

1.

创建一个由新信息构成的中间表。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本，其中包含第 1 步到第 4 步的 SQL 语句：

@mti2_prepare.sql

Rem create intermediate table with some records


CREATE TABLE customers_new AS 
SELECT * 
FROM customers 
WHERE cust_id BETWEEN 2000 AND 5000;

 

2.

禁用对 SALES 表的约束，这是第 3 步所必需的。

ALTER TABLE sales DISABLE CONSTRAINT sales_customer_fk;
3. 从 CUSTOMERS 表中删除一些数据。 
DELETE FROM customers WHERE cust_id BETWEEN 2000 AND 5000;
4.

为特殊的促销信息创建一个空表。

CREATE TABLE customers_special AS
SELECT cust_id, cust_credit_limit 
FROM customers 
WHERE rownum > 1;

 
5. 对多个具有不同表结构的表执行多表插入。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@do_mti2.sql

INSERT /*+ APPEND NOLOGGING */FIRST
WHEN cust_credit_limit >= 4500 THEN
INTO customers
INTO customers_special VALUES (cust_id, cust_credit_limit)
ELSE
INTO customers
SELECT * FROM customers_new;

 
6. 可以查看所插入的内容。执行以下 SQL 脚本：

@control_mti2.sql

SELECT COUNT(*) FROM customers;
SELECT COUNT(*) FROM customers_special;
SELECT MIN(cust_credit_limit) FROM customers_special;
7.

在继续操作前，执行以下 SQL 脚本重置环境：

@reset_mti2.sql


set echo on
REM cleanup and reset

ALTER TABLE sales 
MODIFY CONSTRAINT sales_customer_fk RELY 
ENABLE NOVALIDATE;
DROP TABLE customers_special;
DROP TABLE customers_new;
COMMIT;

返回主题

返回主题列表

使用 Upsert 功能 - SQL MERGE 关键字概述

MyCompany 必须使用其产品数据库中的数据定期更新数据仓库中的产品信息。遗憾的是，尽管您收到了产品更新信息，但却无法区分新信息和更新的信息。因此，必须在数据仓库那边上弄清此问题。

要使用 SQL MERGE 命令有条件地更新或插入数据，执行下列步骤：

1. 为外部产品信息创建外部表（和目录）。
2. 使用 SQL MERGE 命令执行 Upsert。
3. 显示 MERGE 命令的执行计划。
4. 使用两个单独的 SQL 命令执行 Upsert（在没有 MERGE 功能的情况下）

返回主题列表

1. 为外部产品信息创建外部表（和目录）

在创建外部表之前，您需要在数据库中创建一个目录对象，该对象将指向数据文件所在文件系统上的目录。或者，您还可以将记录的坏文件和废弃文件的位置与这些数据文件的位置分开，如以下示例所示。

1.

要创建目录，执行下列步骤。如果您已经完成了高速数据加载和滚动窗口操作，则目录已经存在，因此您可以跳过该步骤。

这些脚本是针对 Linux 系统设置的，并且假设已将上机操作讨论会解压缩到文件夹
/home/oracle/wkdir 中。

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@create_directory.sql

DROP DIRECTORY data_dir;
DROP DIRECTORY log_dir;
Rem *****
Rem CREATE DIRECTORIES
Rem note that security is controlled
Rem (a) with privileges on directory (read/write)
Rem (b) with privileges on external table
Rem *****
CREATE DIRECTORY data_dir AS '/home/oracle/wkdir/';
CREATE DIRECTORY log_dir AS '/home/oracle/wkdir/';
2.

创建外部表时，您要定义两部分信息：

  1. 用于在数据库中表示表的元数据信息
  2. 有关如何从外部文件中提取数据的访问参数定义

创建这些元信息之后，无需初始加载就可以从数据库中访问外部数据。要创建外部表，在登录 SH 模式的 SQL*Plus 会话中执行以下 SQL 脚本：

@create_external_table2.sql

Rem *****
Rem preparation for upsert
Rem file prodDelta.dat, 1.2 MB must
Rem reuse of directory STAGE_DIR
Rem access via external table
Rem *****
DROP TABLE products_delta;
CREATE TABLE products_delta
(
PROD_ID NUMBER(6),
PROD_NAME VARCHAR2(50),
PROD_DESC VARCHAR2(4000),
PROD_SUBCATEGORY VARCHAR2(50),
PROD_SUBCATEGORY_ID NUMBER,
PROD_SUBCATEGORY_DESC VARCHAR2(2000),
PROD_CATEGORY VARCHAR2(50),
PROD_CATEGORY_ID NUMBER,
PROD_CATEGORY_DESC VARCHAR2(2000),
PROD_WEIGHT_CLASS NUMBER(2),
PROD_UNIT_OF_MEASURE VARCHAR2(20),
PROD_PACK_SIZE VARCHAR2(30),
SUPPLIER_ID NUMBER(6),
PROD_STATUS VARCHAR2(20),
PROD_LIST_PRICE NUMBER(8,2),
PROD_MIN_PRICE NUMBER(8,2),
PROD_TOTAL VARCHAR2(13),
PROD_TOTAL_ID NUMBER,
PROD_SRC_ID NUMBER,
PROD_EFF_FROM DATE,
PROD_EFF_TO DATE,
PROD_VALID CHAR(1)
)
ORGANIZATION external
(
TYPE oracle_loader
DEFAULT DIRECTORY data_dir
ACCESS PARAMETERS 
(
RECORDS DELIMITED BY NEWLINE CHARACTERSET US7ASCII
NOBADFILE
NOLOGFILE
FIELDS TERMINATED BY "|" LDRTRIM
)
location
('prodDelta.dat')
REJECT LIMIT UNLIMITED NOPARALLEL;

返回主题

2. 使用 SQL MERGE 命令执行 Upsert

要使用 SQL MERGE 命令执行 upsert，执行以下步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@do_merge_new.sql

Rem *****
Rem MERGE
Rem new and modified product information has arrived
Rem use of external table again
Rem - seemlessly within the MERGE command !!!
Rem *****
MERGE INTO products t
USING products_delta s
ON ( t.prod_id=s.prod_id )
WHEN MATCHED THEN
UPDATE SET t.prod_name=s.prod_name, t.prod_list_price=s.prod_list_price,
t.prod_min_price=s.prod_min_price
WHEN NOT MATCHED THEN
INSERT    (prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id)
VALUES    (s.prod_id, s.prod_name, s.prod_desc, s.prod_subcategory, s.prod_subcategory_id,
s.prod_subcategory_desc, s.prod_category, s.prod_category_id,
s.prod_category_desc,
s.prod_status, s.prod_list_price, s.prod_min_price, s.prod_total,
s.prod_total_id,
s.prod_weight_class, s.prod_pack_size, s.supplier_id)
;
ROLLBACK;

注意：您将执行 ROLLBACK 操作，以便在随后的操作中使用两个 SQL 语句重新执行同一 upsert。

 

返回主题

3. 显示 MERGE 命令的执行计划

要检查 SQL MERGE 语句的执行计划，执行下列操作：

1.

在登录 SH 模式的 SQL*Plus 会话中，运行 explain_merge_new.sql 脚本，或将以下 SQL 语句复制到 SQL*Plus 会话中：

@explain_merge_new.sql

DELETE FROM plan_table;
COMMIT;
EXPLAIN PLAN FOR
MERGE INTO products t
USING products_delta s
ON ( t.prod_id=s.prod_id )
WHEN MATCHED THEN
UPDATE SET t.prod_name=s.prod_name, t.prod_list_price=s.prod_list_price,
t.prod_min_price=s.prod_min_price
WHEN NOT MATCHED THEN
INSERT    (prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id)
VALUES    (s.prod_id, s.prod_name, s.prod_desc, s.prod_subcategory, s.prod_subcategory_id,
s.prod_subcategory_desc, s.prod_category, s.prod_category_id,
s.prod_category_desc,
s.prod_status, s.prod_list_price, s.prod_min_price, s.prod_total,
s.prod_total_id,
s.prod_weight_class, s.prod_pack_size, s.supplier_id)
;
set linesize 140
select * from table(dbms_xplan.display);

 

返回主题

4. 使用两个单独的 SQL 命令执行 Upsert

在不使用 MERGE 语句的情况下，可以组合使用两个 SQL 语句（insert 和 update）或采用过程方式实施此 upsert 功能。下面显示了如何使用两个 SQL 命令完成该业务问题的 upsert 实施。

为演示其实现方法，执行下列步骤：

1.

要利用可更新的连接视图，连接列必须存在主键或唯一键。因此，您无法直接使用外部表。为此，您需要在数据库中创建一个具有相应唯一键约束的中间表。运行以下脚本，创建必需的结构并执行 upsert：

@do_merge_816.sql

CREATE TABLE prod_delta NOLOGGING AS SELECT * FROM products_delta;
CREATE UNIQUE INDEX p_uk ON prod_delta(prod_id);
UPDATE
(SELECT s.prod_id, s.prod_name sprod_name, s.prod_desc sprod_desc,
s.prod_subcategory sprod_subcategory,
s.prod_subcategory_desc sprod_subcategory_desc, s.prod_category sprod_category,
s.prod_category_desc sprod_category_desc, s.prod_status sprod_status,
s.prod_list_price sprod_list_price, s.prod_min_price sprod_min_price,
t.prod_id, t.prod_name tprod_name, t.prod_desc tprod_desc, t.prod_subcategory
tprod_subcategory,
t.prod_subcategory_desc tprod_subcategory_desc, t.prod_category tprod_category,
t.prod_category_desc tprod_category_desc, t.prod_status tprod_status,
t.prod_list_price tprod_list_price, t.prod_min_price tprod_min_price
FROM products t, prod_delta s WHERE s.prod_id=t.prod_id) JV
SET
tprod_list_price =sprod_list_price,
tprod_min_price =sprod_min_price;
INSERT INTO products
(prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id)
SELECT prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id
FROM prod_delta s WHERE NOT EXISTS (SELECT 1 FROM products t WHERE t.prod_id =
s.prod_id);
ROLLBACK;

请注意，当使用 MERGE 语句时，必须执行两个单独的 SQL 语句才能完成同一功能。

 
2.

在登录 SH 模式的 SQL*Plus 会话中，运行 view_explain_merge.sql 脚本查看执行计划，或将以下 SQL 语句复制到 SQL*Plus 会话中：

@view_explain_merge.sql

PROMPT    show the plans
DELETE FROM plan_table;
COMMIT;

EXPLAIN PLAN FOR
  UPDATE
  (SELECT s.prod_id, s.prod_name sprod_name, s.prod_desc sprod_desc, 
s.prod_subcategory sprod_subcategory,
   s.prod_subcategory_desc sprod_subcategory_desc, s.prod_category sprod_category,
   s.prod_category_desc sprod_category_desc, s.prod_status sprod_status, 
s.prod_list_price sprod_list_price, s.prod_min_price sprod_min_price,
   t.prod_id, t.prod_name tprod_name, t.prod_desc tprod_desc, t.prod_subcategory tprod_subcategory,
   t.prod_subcategory_desc tprod_subcategory_desc, t.prod_category tprod_category,
   t.prod_category_desc tprod_category_desc, t.prod_status tprod_status, 
t.prod_list_price tprod_list_price,   t.prod_min_price tprod_min_price
   FROM products t, prod_delta s WHERE s.prod_id=t.prod_id) JV
   SET
   tprod_list_price =sprod_list_price,
   tprod_min_price =sprod_min_price;
set linesize 180
select * from table(dbms_xplan.display);
DELETE FROM plan_table;

COMMIT;


EXPLAIN PLAN FOR
INSERT INTO products
(prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id)
SELECT prod_id, prod_name, prod_desc, prod_subcategory, prod_subcategory_id,
prod_subcategory_desc, prod_category, prod_category_id, prod_category_desc,
prod_status, prod_list_price, prod_min_price, prod_total, prod_total_id,
prod_weight_class, prod_pack_size, supplier_id
FROM prod_delta s WHERE NOT EXISTS (SELECT 1 FROM products t WHERE t.prod_id = s.prod_id);
set linesize 180
select * from table(dbms_xplan.display);
ROLLBACK;
    
...
要利用可更新的连接视图功能，需要将外部表复制到实际的数据库表中，并需要创建连接列的唯一索引。简而言之，这需要更多的操作、更多的空间要求以及更多的处理时间。

返回主题

返回主题列表

 

了解 DML 错误记录功能

Oracle 数据库 10g 第 2 版引入了一个激动人心的新功能，支持对批量 SQL 操作进行记录级别的错误处理。DML 错误记录功能使您可以指定错误记录表（Oracle 将把 DML 操作过程中遇到的错误记录到该表中）的名称，从而扩展了现有的 DML 操作。这使您无论在出现任何错误的情况下都可以完成 DML 操作，并可以在以后的某个时间对错误行采取更正操作。

DML 错误记录表由多个必需的控制列和一组用户定义的列组成。通过使用能够存储目标列潜在错误的数据类型，这些列可以表示 DML 操作的目标表的所有列或列子集。例如，您需要在错误记录表中使用 VARCHAR2 数据类型存储目标表中 NUMBER 列的 TO_NUM 数据类型转换错误。应使用 DBMS_ERRLOG 程序包创建 DML 错误记录表。

1.

创建一个外部表，表示来自源系统的、数据质量明显较差的数据。运行脚本 cr_ext_tab_for_elt.sql。

@cr_ext_tab_for_elt.sql

PROMPT Create an external table pointing to a data set with poor quality
Rem
DROP TABLE sales_activity_direct;

CREATE TABLE sales_activity_direct
(
  PROD_ID NUMBER,
  CUST_ID NUMBER,
  TIME_ID CHAR(20),
  CHANNEL_ID CHAR(2),
  PROMO_ID NUMBER,
  QUANTITY_SOLD NUMBER(3),
  AMOUNT_SOLD CHAR(50)
)
  ORGANIZATION external
(
  TYPE oracle_loader
  DEFAULT DIRECTORY data_dir
  ACCESS PARAMETERS
(
  RECORDS DELIMITED BY NEWLINE CHARACTERSET US7ASCII
  BADFILE log_dir:'sh_sales2.bad'
  LOGFILE log_dir:'sh_sales2.log_xt'
  FIELDS TERMINATED BY "|" LDRTRIM
)
  location
(
  'sales_activity_direct.dat'
)
  )REJECT LIMIT UNLIMITED NOPARALLEL;

 
2. 创建一个虚拟表以避免干扰现有表。运行脚本 cr_tab_for_elt.sql。

@cr_tab_for_elt.sql

PROMPT create a second sales table to being used by the insert
DROP TABLE sales_overall;
CREATE TABLE sales_overall as select * from sales where 1=0;
3. 要使用新的 DML 错误记录功能跟踪错误，需要一个错误记录表。强烈
建议您使用提供的程序包 DBMS_ERRLOG。运行脚本 cr_elt.sql。

@cr_elt.sql

PROMPT Create the DML error logging table with DEFAULT name
PROMPT It is highly advised to use the ORCL-provided pkg to create the
PROMPT DML Error Logging table
Rem
DROP TABLE err$_sales_overall;
exec dbms_errlog.create_error_log('sales_overall');

注意必需的控制列以及针对要跟踪的列的不同数据类型。数据
类型必须是目标数据类型的超集才能正确地跟踪错误，例如，对于数字目标列应使用非数字值
。

DESCRIBE err$_sales_overall

 
4.

尝试将外部文件中的数据加载到目标表中。错误
记录功能的默认行为是设置一个值为 0 的 REJECT LIMIT。出现错误时，DML 操作将失败
，并将引发错误的第一个记录存储在错误记录表中。运行脚本 ins_elt_1.sql 查看此行为。

@ins_elt_1.sql

PROMPT First insert attempt , DEFAULT reject limit 0
PROMPT also, the error message that comes back is the one of error #1
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_test1' );
commit;


PROMPT As you can see, nothing is inserted in the target table, but ONE row in the error
logging table
select count(*) from sales_overall;
select count(*) from err$_sales_overall;
delete from err$_sales_overall;
commit;

从 DML 的角度而言，您需要确保事务要么成功要么失败。因此，从一般角度而言，您希望 DML 操作在任何情况下均成功（这将转换为一个无限制的 reject limit），或者希望它在出现错误的情况下失败 (reject limit 0)。由于所选择的任意数字在某种程度上没有意义，因此决定将默认的 reject limit 设置为 0；如果决定容许特定数量的错误，则在特定情况下允许多少个错误是一个纯粹的业务决策。

 
5.

通过将 REJECT LIMIT 设置为 10 个记录再次尝试插入。如果出现的错误数大于 10，则 DML 操作将失败，您将在错误记录表中发现 11 个记录。运行脚本 ins_elt_2.sql 以查看此行为。

@ins_elt_2.sql

SET ECHO OFF

PROMPT First insert attempt , DEFAULT reject limit 10
PROMPT Note that the error message that comes back is the one of error #11


INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_test2' ) REJECT LIMIT 10;
commit;


PROMPT no rows in target; error count+1 in DML error logging table
select count(*) from sales_overall;
select count(*) from err$_sales_overall;
delete from err$_sales_overall;
commit;

此插入的错误数大于 10，这意味着该数据的质量很差。

 
6.

将数据放入表中，并计算遇到的错误数。运行脚本 ins_elt_3.sql。

@ins_elt_3.sql

PROMPT Reject limit unlimited will succeed
Rem ... as long as you do not run into one of the current limitations ...
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) REJECT LIMIT UNLIMITED;
commit;


PROMPT finally ...
select count(*) from sales_overall;
select count(*) from err$_sales_overall;

 
7.

出现了大量的错误！查看错误记录表，以对出现的错误有一个更好的了解。运行脚本 sel_elt_1.sql。

@sel_elt_1.sql

PROMPT Please recognize the subtle difference between ERROR MESSAGE ONLY and -
ERROR MESSAGE TEXT


Rem Therefore we enforce to store both
set linesize 80

select distinct ora_err_number$ from err$_sales_overall;
select distinct ora_err_number$, ora_err_mesg$ from err$_sales_overall;

请注意 Oracle 错误消息与 Oracle 错误文本之间的细微差别。很多情况下，错误
文本提供了有助于分析问题的附加信息，因此错误消息和错误文本都是错误记录表的必要控制列。

 
8.

时至今日，仍有一些与错误记录功能相关的限制。所有限制都涉及到在延迟优化中执行索引维护的情况。未来版本的 Oracle 将解决这些限制。运行脚本 elt_limit.sql。

@elt_limit.sql 
PROMPT discuss a case with limitation
truncate table sales_overall;
alter table sales_overall add constraint pk_1 primary key (prod_id, cust_id,time_id,
channel_id, promo_id);

PROMPT works fine as before, errors get re-routed
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) REJECT LIMIT UNLIMITED;
commit;

select count(*) from sales_overall;
select count(*) from err$_sales_overall;
delete from err$_sales_overall;
commit;
PROMPT case when deferred constraint check (UNIQUE INDEX) leads to error
Rem unique index maintenance is a delayed operation that cannot be cuaght
Rem on a perrecord  base as of today.Planned for a future release.
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) REJECT LIMIT UNLIMITED;
commit;

select count(*) from sales_overall;
select count(*) from err$_sales_overall;
commit;
9.

DML 错误记录将记录 DML 操作过程中出现的任何类型的错误。但无法捕获在 DML 操作之前作为执行计划中 SQL 语句组成部分发生的任何类型的错误。来看以下脚本 ins_elt_4.sql 中的 DML 操作。

@ins_elt_4.sql 

PROMPT INTERESTING CASE.
PROMPT The DML Error Logging can only catch errors that happen at DML TIME, but not errors
PROMT that happen as part of the SQL statement
Rem the TO_DATE conversion is in the insert portion, so that we catch the error
Rem this is default view merging of ORCL 
alter table sales_overall drop constraint pk_1;
truncate table sales_overall;
INSERT /*+ APPEND NOLOGGING PARALLEL */
INTO sales_overall
SELECT prod_id, cust_id,
TO_DATE(time_id,'DD-mon-yyyy'),
channel_id, promo_id, quantity_sold, amount_sold
FROM sales_activity_direct
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) 
REJECT LIMIT UNLIMITED;
commit

请注意，SELECT 语句将 TO_DATE() 函数应用于列 sales.activity_direct.time_id。Oracle 通过将此转换函数向上传递给插入操作在内部优化了错误处理，以确保捕获任何潜在的错误。正如您早先看到的，DML 操作成功完成。

查看该计划时，您将了解到所执行的错误处理的优化类型。运行脚本 xins_elt_4.sql。

@xins_elt_4.sql 

SET LINESIZE 140

explain plan for
INSERT /*+ APPEND NOLOGGING PARALLEL */
INTO sales_overall
SELECT * FROM
( SELECT prod_id, cust_id,
TO_DATE(time_id,'DD-mon-yyyy'),
channel_id, promo_id, quantity_sold, amount_sold
FROM sales_activity_direct )
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) 
REJECT LIMIT UNLIMITED;

select * from table(dbms_xplan.display(null,null,'ALL'));

该计划表明 TO_DATE() 函数向上传递给表示为 'LOAD AS
SELECT' 行源的插入操作（它是一个直接路径插入）

 
10.

再次运行同一 DML 操作，但强制规定不能将 TO_DATE() 转换传递给 DML 操作。通过使用一个带有 NO_MERGE 提示的视图结构可以实现此目的。等价的 SQL SELECT 语句本身失败，您可以通过运行脚本 sel_ins_elt_5.sql 看到此情况的发生。

@sel_ins_elt_5.sql 

PROMPT The equivalent SQL statement will fail if the predicate is artificially kept -
within a lower query block

SELECT prod_id, cust_id,
TO_DATE(time_id,'DD-mon-yyyy'),
channel_id, promo_id, quantity_sold, amount_sold
FROM sales_activity_direct;

DML 操作将因为同一错误而失败。运行脚本 ins_elt_5.sql。

@ins_elt_5.sql 
PROMPT The same is true when errors are happening on a row source level inside the -
SELECT block
PROMPT We cannot catch such an error 


truncate table sales_overall;
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM
( SELECT /*+ NO_MERGE */ prod_id, cust_id,
TO_DATE(time_id,'DD-mon-yyyy'),
channel_id, promo_id, quantity_sold, amount_sold
FROM sales_activity_direct )
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) 
REJECT LIMIT UNLIMITED;
commit;

该计划表明转换发生在视图内部，因此出现错误。运行脚本 xins_elt_5.sql 以查看此行为：

@xins_elt_5.sql 

explain plan for
INSERT /*+ APPEND NOLOGGING PARALLEL */ INTO sales_overall
SELECT * FROM
( SELECT /*+ NO_MERGE */ prod_id, cust_id,
TO_DATE(time_id,'DD-mon-yyyy'),
channel_id, promo_id, quantity_sold, amount_sold
FROM sales_activity_direct )
LOG ERRORS INTO err$_sales_overall ( 'load_20040802' ) 
REJECT LIMIT UNLIMITED;

select * from table(dbms_xplan.display(null,null,'ALL'));

返回主题

返回主题列表

表函数概述

在 ETL 过程中，从源系统中提取的数据在加载到数据仓库之前，将通过一系列转换。复杂转换是使用 PL/SQL 以过程方式在数据库的外部或内部实施的。如果转换结果太大，无法存储到内存中，则必须通过将它们物化到数据库表或平面文件中来临时存储它们。随后，该数据作为序列中下一个转换的输入进行读取和处理。Oracle 表函数支持对使用 PL/SQL、C 或 Java 实施的此类转换进行管道化和并行执行。

表函数定义为一个可以生成一组行作为输出的函数；此外，Oracle 表函数还可以接受一组行作为输入。这些行集在子集中以迭代方式进行处理，从而使管道机制能够在第一个操作完成前将这些子集结果从一个转换传送到下一个转换。此外，表函数还可以通过并行方式进行透明处理，这相当于表扫描或排序等 SQL 操作。不再需要临时存储表。

由于将表函数用作复杂转换的一部分发挥了强大的功能并具有多面性，因此您将只需检查表函数的基本实施问题。为此，执行下列步骤：

1. 为表函数设置基本对象。
2. 执行非管道化的表函数，并返回一个记录数组。
3. 执行管道化的表函数。
4. 透明地并行执行表函数。
5. 使用独立的 DML 执行表函数。
6. 通过多个表函数执行无缝的传送。

 

1. 为表函数设置基本对象

表函数可以生成一组行作为输出，并可以接受一组行作为输入。因此，您必须定义一个记录类型和一个记录集合来定义输入和输出；这些类型既可以是强类型的（即，在其执行之前进行了良好的定义），也可以是弱类型的（这意味着在运行时之前没有固定的记录格式）。您将同时检查这两个变体。

要设置表函数对象，执行下列步骤：

1.

为示例定义对象（记录）类型。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@setup_tf_record.sql

SET ECHO ON
PROMPT object_types
CREATE TYPE product_t AS OBJECT (
prod_id NUMBER(6)
, prod_name VARCHAR2(50)
, prod_desc VARCHAR2(4000)
, prod_subcategory VARCHAR2(50)
, prod_subcategory_desc VARCHAR2(2000)
, prod_category VARCHAR2(50)
, prod_category_desc VARCHAR2(2000)
, prod_weight_class NUMBER(2)
, prod_unit_of_measure VARCHAR2(20)
, prod_pack_size VARCHAR2(30)
, supplier_id NUMBER(6)
, prod_status VARCHAR2(20)
, prod_list_price NUMBER(8,2)
, prod_min_price NUMBER(8,2)
);

这是您将在示例中使用的基本对象记录类型。

 

2. 为示例定义对象（集合）类型。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@setup_tf_collection.sql

CREATE TYPE product_t_table AS TABLE OF product_t;
/

该类型表示表函数将返回的记录集的结构。
3. 为 REF CURSOR 类型定义一个程序包。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@setup_tf_package.sql

Rem package of all cursor types
Rem handle the input cursor type and the output cursor collection type


CREATE OR REPLACE PACKAGE cursor_PKG as
TYPE product_t_rec IS RECORD (prod_id NUMBER(6) 
, prod_name VARCHAR2(50)
, prod_desc VARCHAR2(4000)
, prod_subcategory VARCHAR2(50)
, prod_subcategory_desc VARCHAR2(2000)
, prod_category VARCHAR2(50)
, prod_category_desc VARCHAR2(2000)
, prod_weight_class NUMBER(2)
, prod_unit_of_measure VARCHAR2(20)
, prod_pack_size VARCHAR2(30)
, supplier_id NUMBER(6)
, prod_status VARCHAR2(20)
, prod_list_price NUMBER(8,2)
, prod_min_price NUMBER(8,2));


TYPE product_t_rectab IS TABLE OF product_t_rec;
TYPE strong_refcur_t IS REF CURSOR RETURN product_t_rec;
TYPE refcur_t IS REF CURSOR;
END;

 
4.

使用独立的 DML 为表函数示例创建一个日志表。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@cre_ope_table.sql

CREATE TABLE obsolete_products_errors 
(prod_id NUMBER, msg VARCHAR2(2000));

该表将用于在其执行过程中分发某些数据的表函数示例。

由于时间方面的限制，此处只对表函数的 PL/SQL 实施进行了介绍。有关更多信息，请参见“Data Cartridge 开发人员指南”，以了解接口实施的详细信息。

表函数向随后的操作返回一组记录。 The result set can be either delivered en-block (the default PL/SQL behavior) or pipelined – increments of the result set are streamed to the subsequent operation as soon as they are produced.

对于 PL/SQL 中表函数的实现，PL/SQL 引擎控制以增量方式返回的数组的大小。对于用 C 或 Java 编写的表函数的接口实现，由程序员定义增量式返回集。

返回主题

2. 执行非管道化的表函数，并返回一个记录数组

以下示例显示了一个简单的表函数，它以完整的数组形式返回其非管道化的结果集。该表函数滤掉产品类别“Electronics”的所有记录并返回所有其他记录。请注意，输入 REF CURSOR 是弱类型的，这意味着它可以为任何有效的 SQL 语句。表函数代码负责处理它的输入（可能基于其他输入变量）。

要执行非管道化表函数，执行下列步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，运行 create_tf_1.sql，或将下列 SQL 语句复制到 SQL*Plus 会话中：

@create_tf_1.sql

PROMPT SIMPLE PASS-THROUGH, FILTERING OUT obsolete products without
product_category 'Electronics'
Rem uses weakly typed cursor as input
CREATE OR REPLACE FUNCTION obsolete_products(cur cursor_pkg.refcur_t) RETURN
product_t_table
IS
prod_id NUMBER(6);
prod_name VARCHAR2(50);
prod_desc VARCHAR2(4000);
prod_subcategory VARCHAR2(50);
prod_subcategory_desc VARCHAR2(2000);
prod_category VARCHAR2(50);
prod_category_desc VARCHAR2(2000);
prod_weight_class NUMBER(2);
prod_unit_of_measure VARCHAR2(20);
prod_pack_size VARCHAR2(30);
supplier_id NUMBER(6);
prod_status VARCHAR2(20);
prod_list_price NUMBER(8,2);
prod_min_price NUMBER(8,2);
sales NUMBER:=0;
objset product_t_table := product_t_table();
i NUMBER := 0;
BEGIN
LOOP
-- Fetch from cursor variable
FETCH cur INTO prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price;
EXIT WHEN cur%NOTFOUND; -- exit when last row is fetched
-- Category Electronics is not meant to be obsolete and will be suppressed
IF prod_status='obsolete' AND prod_category !='Electronics' THEN
-- append to collection
i:=i+1;
objset.extend;
objset(i):=product_t( prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price);
END IF;
END LOOP;
CLOSE cur;
RETURN objset;
END;
/
2.

无法透明地从定义的表函数 obsolete_products 中执行 SELECT 操作。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@use_tf_1.sql

SELECT DISTINCT UPPER(prod_category), prod_status
FROM TABLE(obsolete_products(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products)));

请注意，游标变量的输入 SELECT 列表必须匹配表函数的定义（本示例中为 pkg.refcur_t）。如果可以确保表定义（本示例中为 products）匹配该输入游标，则可以使用 SELECT * FROM TABLE 表示法简化该语句。

返回主题

3. 执行管道化的表函数

以下示例显示了与上一个示例相同的滤掉操作。但与第一个示例不同的是，该表函数返回它的管道化结果集，这意味着以增量方式返回结果集。请注意，输入 REF CURSOR 在此处为强类型，这意味着它只能是一个返回 product_type_record 记录类型的 SQL 语句。

要执行管道化表函数，执行下列步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@create_tf_2.sql

Rem same example, pipelined implementation
Rem strong ref cursor (input type is defined)
Rem a table without a strong typed input ref cursor 
Rem cannot be parallelized
CREATE OR REPLACE FUNCTION obsolete_products_pipe(cur cursor_pkg.strong_refcur_t)
RETURN product_t_table
PIPELINED
PARALLEL_ENABLE (PARTITION cur BY ANY) IS
prod_id NUMBER(6);
prod_name VARCHAR2(50);
prod_desc VARCHAR2(4000);
prod_subcategory VARCHAR2(50);
prod_subcategory_desc VARCHAR2(2000);
prod_category VARCHAR2(50);
prod_category_desc VARCHAR2(2000);
prod_weight_class NUMBER(2);
prod_unit_of_measure VARCHAR2(20);
prod_pack_size VARCHAR2(30);
supplier_id NUMBER(6);
prod_status VARCHAR2(20);
prod_list_price NUMBER(8,2);
prod_min_price NUMBER(8,2);
sales NUMBER:=0;
BEGIN
LOOP
-- Fetch from cursor variable
FETCH cur INTO prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price;
EXIT WHEN cur%NOTFOUND; -- exit when last row is fetched
IF prod_status='obsolete' AND prod_category !='Electronics' THEN
PIPE ROW (product_t( prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price));
END IF;
END LOOP;
CLOSE cur;
RETURN;
END;
/
2.

您现在可以从定义的表函数 obsolete_products 中透明地执行 SELECT 操作。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@use_tf_2.sql

SELECT DISTINCT prod_category, DECODE(prod_status,'obsolete','NO LONGER 
AVAILABLE','N/A')
FROM TABLE(obsolete_products_pipe(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products)));

 

返回主题

4. 透明地并行执行表函数

表函数可以透明地并行执行，而不需要任何特殊的干预或设置。在创建表函数的过程中必须执行的唯一声明就是通知 SQL 引擎在并行执行表函数方面是否存在任何限制或规则。

假设有一个根据某些产品类别属性处理复杂聚合操作的表函数。如果要并行处理该操作，则必须确保从同一并行执行 Slave 中处理具有相同产品类别属性的所有记录，以便聚合操作涵盖属于同一组的所有记录。这意味着，在并行处理过程中，必须对并行处理使用一个分发规则，这将确保如上所述的正确分发。

将该并行分发规则定义为表函数标题的一部分，例如：

PARALLEL_ENABLE (PARTITION cur BY ANY) IS

您只需使用所显示的语法即可实现表函数的并行执行。

要并行执行前面定义的某个表函数，执行下列步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@pq_session.sql

SELECT * 
FROM v$pq_sesstat 
WHERE statistic in ('Queries Parallelized','Allocation Height');
2.

现在，您可以并行使用某个表函数。使用 PARALLEL 提示对 REF CURSOR 实施并行执行。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@parallel_tf.sql

SELECT DISTINCT prod_category,
DECODE(prod_status,'obsolete','NO LONGER AVAILABLE','N/A')
FROM TABLE(obsolete_products_pipe(
CURSOR(SELECT /*+ PARALLEL(a,4)*/
prod_id, prod_name, prod_desc, prod_subcategory, prod_subcat_desc,
prod_category, prod_cat_desc, prod_weight_class, prod_unit_of_measure,
prod_pack_size, supplier_id,prod_status, prod_list_price,prod_min_price 
FROM products a)));

 
3.

重新运行此 SQL 脚本以检查并行查询的状态：

@pq_session.sql

SELECT * 
FROM v$pq_sesstat 
WHERE statistic in ('Queries Parallelized','Allocation Height');

 

返回主题

5. 使用独立 DML 执行表函数

尽管表函数是单个原子事务的一部分，但它还提供了其他功能，用于将数据分发到独立事务作用 域中的其他表。可以通过各种方式使用该功能，如将它用于异常或进度日志记录，或分发要由其他独立转换使用的数据子集。以下示例将一些“异常”信息记录到了 OBSOLETE_PRODUCTS_ERRORS 表中：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@create_tf_3.sql

CREATE OR REPLACE FUNCTION obsolete_products_dml(cur cursor_pkg.strong_refcur_t,
prod_cat varchar2 DEFAULT 'Electronics') RETURN product_t_table
PIPELINED
PARALLEL_ENABLE (PARTITION cur BY ANY) IS
PRAGMA AUTONOMOUS_TRANSACTION;
prod_id NUMBER(6);
prod_name VARCHAR2(50);
prod_desc VARCHAR2(4000);
prod_subcategory VARCHAR2(50);
prod_subcategory_desc VARCHAR2(2000);
prod_category VARCHAR2(50);
prod_category_desc VARCHAR2(2000);
prod_weight_class NUMBER(2);
prod_unit_of_measure VARCHAR2(20);
prod_pack_size VARCHAR2(30);
supplier_id NUMBER(6);
prod_status VARCHAR2(20);
prod_list_price NUMBER(8,2);
prod_min_price NUMBER(8,2);
sales NUMBER:=0;
BEGIN
LOOP
-- Fetch from cursor variable
FETCH cur INTO prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price;
EXIT WHEN cur%NOTFOUND; -- exit when last row is fetched
IF prod_status='obsolete' THEN
IF prod_category=prod_cat THEN
INSERT INTO obsolete_products_errors VALUES
(prod_id, 'correction:category '||UPPER(prod_cat)||' still available');
COMMIT;
ELSE
PIPE ROW (product_t( prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price));
END IF;
END IF;
END LOOP;
CLOSE cur;
RETURN;
END;
/
2.

现在，您可以从该表函数中进行选择并检查日志表。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@use_tf_3a.sql

TRUNCATE TABLE obsolete_products_errors;
SELECT DISTINCT prod_category, prod_status FROM TABLE(obsolete_products_dml(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products)));
SELECT DISTINCT msg FROM obsolete_products_errors;

 
3.

使用不同的输入参数再次从该表函数中进行选择，并检查日志表。

@use_tf_3b.sql

TRUNCATE TABLE obsolete_products_errors;
SELECT DISTINCT prod_category, prod_status FROM TABLE(obsolete_products_dml(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products),'Photo'));
SELECT DISTINCT msg FROM obsolete_products_errors;

您会看到，根据输入变量的不同，表函数将采用不同的方法处理输入行，从而提供一个不同的结果集并将不同的信息插入到 obsolete_products_error 中。

返回主题

6. 通过多个表函数执行无缝流

除了在 SQL 语句中透明地使用以及能够被并行处理以外，表函数的最大好处之一是能够相互进行调用。此外，表函数可以在任何 SQL 语句中使用并成为所有类型的 DML 语句的输入。

要了解如何执行此操作，执行下列步骤：

1.

该 SQL 语句使用两个相互嵌套的表函数。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@use_tf_stream.sql

SELECT DISTINCT prod_category, prod_status
FROM TABLE(obsolete_products_dml(CURSOR(SELECT *
FROM TABLE(obsolete_products_pipe(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products))))));
SELECT COUNT(*) FROM obsolete_products_errors;
SELECT DISTINCT msg FROM obsolete_products_errors;
set echo off
2.

现在，您可以将该表函数用作 CREATE TABLE AS SELECT 命令的输入。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@CTAS_tf.sql

CREATE TABLE PIPE_THROUGH AS
SELECT DISTINCT prod_category, prod_status
FROM TABLE(obsolete_products_dml(CURSOR(SELECT *
FROM TABLE(obsolete_products_pipe(
CURSOR(SELECT prod_id, prod_name, prod_desc, prod_subcategory,
prod_subcategory_desc,
prod_category, prod_category_desc, prod_weight_class,
prod_unit_of_measure,
prod_pack_size, supplier_id, prod_status, prod_list_price,
prod_min_price
FROM products))))));

 
3.

在继续学习本教程的下个部分之前，您需要执行以下脚本来清理环境：

@cleanup_tf.sql

DROP TABLE obsolete_products_errors;
DROP TABLE pipe_through;
DROP FUNCTION obsolete_products;
DROP FUNCTION obsolete_products_pipe;
DROP FUNCTION obsolete_products_dml;

 

返回主题

返回主题列表

使用同步更改数据捕获 (CDC) 跟踪和使用增量式源更改

数据仓库涉及从一个或多个源数据库中提取关系数据，然后将其传输到数据仓库进行分析。更改数据捕获 (Change Data Capture) 快速识别并仅处理已经更改的数据而非整个表，并使更改数据可供将来使用。

如果没有更改数据捕获，那么数据库提取将是一个烦琐的过程：将整个表内容移动到平面文件中，然后将这些文件加载到数据仓库中。此即席方法在很多方面开销很大。

更改数据捕获并不依赖中间平面文件将数据临时存储在关系数据库的外部。它捕获对用户表执行 INSERT、UPDATE 和 DELETE 操作而生成的更改数据。更改数据随后存储到名为更改表的数据库对象中，然后通过控制方法将更改数据提供给应用程序。

发布和预订模型

大多数更改数据捕获系统都有一个发布方，负责捕获和发布任意数量的 Oracle 源表的更改数据。可能有多个访问更改数据的预订方。更改数据捕获提供了一个 PL/SQL 程序包来完成发布和预订任务。

发布方

发布者通常是数据库管理员 (DBA)，他负责创建和维护构成更改数据捕获系统的模式对象。发布方执行以下任务：

确定数据仓库应用程序要从中捕获更改数据的关系表（称作源表）。

使用 Oracle 提供的程序包 DBMS_LOGMNR_CDC_PUBLISH 设置系统，以便从一个或多个源表中捕获数据。
以更改表的形式发布更改数据。
允许对预订方进行受控制的方法，方法是使用 SQL GRANT 和 REVOKE 语句向用户和角色授予和调用更改表的 SELECT 权限。

预订方

预定方（通常是应用程序）是已发布的更改数据的使用者。预定方预定源表中的一组或多组列。预定方执行以下任务：

使用 Oracle 提供的程序包 DBMS_LOGMNR_CDC_SUBSCRIBE 预定源表，对已发布的更改数据进行受控制的访问，以便对这些数据进行分析。

当预订方准备接收一组更改数据时，扩展预订窗口并创建一个新的预订方视图。
使用 SELECT 语句从预订方视图中检索更改数据。
在处理完更改块后，删除预订方视图并清除预订窗口。
当预订方不再需要它的更改数据时删除预订。

有关对更改数据捕获体系结构更详细的介绍，请参阅“Oracle 数据仓库指南”。

要了解有关该主题的更多信息，执行以下步骤：

1. 使用同步 CDC 跟踪所有增量式更改。
2. 创建一个更改表。
3. 预订更改集以及相关的所有源表列。
4. 激活预订并扩展预定窗口。
5. 了解如何随时间的推移来处理新环境。
6. 运行发布方。
7. 删除已使用的更改视图并清除预订窗口。
8. 清理 CDC 环境。

返回主题

返回主题列表

1.使用同步 CDC 跟踪所有增量式更改

您将通过执行以下步骤使用同步 CDC 跟踪表 PRODUCTS 的所有增量式更改（出于演示目的，发布方和预订方将为同一数据库用户。）：

要了解如何执行此操作，执行下列步骤：

1.

首先创建一个在其中应用所有更改的新中间表。在登录 SH 模式的 SQL*Plus 会话中，运行 cr_cdc_target.sql，或将以下 SQL 语句复制到 SQL*Plus 会话中：

@cr_cdc_target.sql

CREATE TABLE my_price_change_Electronics
(prod_id number, prod_min_price number, prod_list_price number, when date);


2.

如果以前从未使用过更改数据捕获，那么除了同步更改集 SYNC_SET 和同步更改源 SYNC_SOURCE 以外，您将看不到与 CDC 相关的任何对象。这两个对象是在创建数据库的过程中创建的。

@show_cdc_env1.sql

SELECT * FROM change_sources;
PROMPT see the change tables
Rem shouldn't show anything
SELECT * FROM change_tables;
PROMPT see the change sets SYNC_SET
SELECT decode(to_char(end_date,'dd-mon-yyyy hh24:mi:ss'),null,'No end date set.')
end_date,
decode(to_char(freshness_date,'dd-mon-yyyy hh24:mi:ss'),null,'No freshness
date set.') freshness_date
FROM change_sets WHERE set_name='SYNC_SET';


返回主题

2.创建更改表

所有更改都存储在更改表中。更改表与源表保持一对一的相关性；它们由多个固定的元数据列和一组动态列（相当于已标识的相关源列）组成。

1.

使用提供的程序包 DBMS_CDC_PUBLISH 创建更改表。

请注意，旧程序包名称 DBMS_CDC_LOGMNR_PUBLISH 是一个指向 DBMS_CDC_PUBLISH 的同义词，并按原样维护以实现向后兼容性。

@cr_cdc_ct.sql

rem create a change table within the change set SYNC_SET.
Rem details on parameters see doc
PROMPT ***10g***
PROMPT NOTE THAT DBMS_LOGMNR_CDC_* are synonyms for the DBMS_CDC_* packages and are only around
PROMPT for backwards compatibility
begin
DBMS_CDC_PUBLISH.CREATE_CHANGE_TABLE (OWNER => 'SH', -
CHANGE_TABLE_NAME => 'PROD_price_CT',
CHANGE_SET_NAME => 'SYNC_SET',
SOURCE_SCHEMA => 'SH',
SOURCE_TABLE => 'PRODUCTS',
COLUMN_TYPE_LIST => 'prod_id number(6), prod_min_price number(8,2),
prod_list_price number(8,2)',
CAPTURE_VALUES => 'both',
RS_ID => 'y',
ROW_ID => 'n',
USER_ID => 'n',
TIMESTAMP => 'n',
OBJECT_ID => 'n',
SOURCE_COLMAP => 'y',
TARGET_COLMAP => 'y',
OPTIONS_STRING => null);
end;
/

该脚本创建一个名为 PROD_PRICE_CT 的更改表和必需的触发器框架，以跟踪随后对 PRODUCTS 进行的所有更改。请注意，跟踪是作为原子 DML 操作的一部分针对 PRODUCTS 进行的。
2. 描述更改表 PROD_PRICE_CT： 
set linesize 120
desc prod_price_ct

参见静态元数据列以及列 PROD_ID、PROD_MIN_PRICE 和 PROD_LIST_PRICE（与源表 PRODUCTS 等价）。

 
3.

要查看系统中已发布的源表和更改表的元数据，执行以下 SQL 命令：

@show_cdc_env2.sql

prompt see the published source tables
SELECT * FROM dba_source_tables;
prompt see published source columns
SELECT source_schema_name,source_table_name, column_name
FROM dba_source_tab_columns;
prompt see the change tables
SELECT * FROM change_tables;

返回主题

3. 预订更改集和相关的所有源表列

可以预订更改集，以对发布的更改数据进行受控制的访问，从而进行分析。请注意，预定句柄的逻辑实体是一个更改集，而非更改表。更改集可以由多个更改表组成，并确保其所有更改表在逻辑上保持一致。预订更改集后，必须预订相关的所有源表列。

1.

必须获取一个在整个会话中使用的唯一预订句柄，并通知系统您所关注的列。您将使用现有的更改集 SYNC_SET。该功能由程序包 DBMS_CDC_SUBSCRIBE 提供。

@subs_cdc_ct.sql

variable subname varchar2(30)
begin
:subname := 'my_subscription_no_1';
DBMS_CDC_SUBSCRIBE.CREATE_SUBSCRIPTION (
CHANGE_SET_NAME => 'SYNC_SET',
DESCRIPTION => 'Change data PRODUCTS for ELECTRONICS',
SUBSCRIPTION_name => :subname);
END;
/
PRINT subname
2. 可以查询有关预订的元信息。

@show_cdc_env3.sql

SELECT handle, description,
decode(status,'N','Not activated.','Activated.') status,
earliest_scn, latest_scn, decode(last_purged,null,'Never purged.',
to_char(last_purged,'dd-mon-yyyy hh24:mi:ss')) last_purged,
decode(last_extended, null,'Never extended.',
to_char(last_extended,'dd-mon-yyyy hh24:mi:ss')) last_extended
FROM 
user_subscriptions;
SELECT * FROM user_subscribed_tables;
SELECT source_table_name, column_name 
FROM dba_subscribed_columns;

 

3.

创建预订后，为该预订设置一个更改视图。Oracle 数据库 10g 对更改视图的完成处理进行了增强：可以命名更改表，或不必删除更改视图即可扩展和/或清除预订方窗口。

@cr_cdc_rv.sql

rem now subscribe to a single source table and columns of interest.
rem A subscription can contain one or more tables from the same change set.
PROMPT ***10g***
Rem use new interface to name subscriber view explictly !!
variable view_name varchar2(30);
BEGIN
:view_name := 'my_prod_price_change_view';
DBMS_CDC_SUBSCRIBE.SUBSCRIBE (
SUBSCRIPTION_NAME => :subname,
SOURCE_SCHEMA => 'sh',
SOURCE_TABLE => 'products',
COLUMN_LIST => 'prod_id, prod_min_price, prod_list_price',
subscriber_view => :view_name );
END;
/
PROMPT ***10g***
Rem The view is created automatically
desc MY_PROD_PRICE_CHANGE_VIEW

到目前为止，您已经完成了设置 CDC 环境所必需的步骤：

在发布方站点上，您为 CDC 准备了源系统。

在预订方站点上，您识别并预订了相关的所有源表（和列），并定义/创建了更改视图。

现在，您准备使用 CDC。

返回主题

4.激活预订并扩展预订窗口

设置发布和预订框架后的第一步是激活预订并扩展预订窗口，以查看自上次扩展预订窗口或激活预订以来进行的所有更改。

1.

激活并扩展预订窗口。您还将查看元数据。

@sub_cdc1.sql

rem now activate the subscription since we are ready to receive 
rem change data the ACTIVATE_SUBSCRIPTION procedure sets 
rem subscription window to empty initially
rem At this point, no additional source tables can be added to the 
rem subscription

EXEC DBMS_LOGMNR_CDC_SUBSCRIBE.ACTIVATE_SUBSCRIPTION -
(SUBSCRIPTION_name => 'my_subscription_no_1')

rem now recheck the subscriptions and see that it is active
rem the view is still not created ...

SELECT handle, description,
decode(status,'N','Not activated.','Activated.') status, earliest_scn, latest_scn,
decode(last_purged,null,'Never purged.', 
to_char(last_purged,'dd-mon-yyyy hh24:mi:ss')) last_purged,
decode(last_extended, null,'Never extended.',
to_char(last_extended,'dd-mon-yyyy hh24:mi:ss')) last_extended
from user_subscriptions;
2. 您现在应用于源表 PRODUCTS 的任何更改都反映在更改表中。更改通过源表上的触发器进行透明地维护。

@dml_cdc1.sql

Rem now do some changes

UPDATE products
SET prod_list_price=prod_list_price*1.1
WHERE prod_min_price > 100;
COMMIT;
Rem you will see entries in the change table
Rem note that we have entries for the old and the new values

SELECT Count(*) FROM prod_price_ct;

请注意，每个源行都有两个记录，一个表示旧值，另一个则是为预订列设置的新值。

还应注意的是，切勿使用更改表本身标识对源表进行的更改。使用将为您创建的预订方视图。这是唯一一种确保将每个更改只传递给预订应用程序一次的受支持的方法。
3.

与更改表不同，更改视图不显示任何记录。这是因为您是在扩展了预订窗口之后应用了 DML 操作。因此，您一定不会看到任何内容。

Rem the view does not show anything
select count(*) from MY_PROD_PRICE_CHANGE_VIEW;

您现在准备通过扩展预订窗口来使用更改。

@ext_cdc_sub1.sql

rem now set upper boundary (high-water mark) for the subscription window
rem At this point, the subscriber has created a new window that begins
rem where the previous window ends.
rem The new window contains any data that was added to the change table.

EXEC DBMS_CDC_SUBSCRIBE.EXTEND_WINDOW -
(SUBSCRIPTION_NAME => 'my_subscription_no_1');
        
rem now recheck the subscriptions and see that it is active
SELECT handle, description,
decode(status,'N','Not activated.','Activated.') status,
earliest_scn, latest_scn,
decode(last_purged,null,'Never purged.',
to_char(last_purged,'dd-mon-yyyy hh24:mi:ss')) last_purged,
decode(last_extended, null,'Never extended.',
to_char(last_extended,'dd-mon-yyyy hh24:mi:ss')) last_extended
from user_subscriptions;
Rem ... and you will see data
SELECT count(*) FROM my_prod_price_change_view;
4.

现在，可以从这个系统生成的更改视图中进行选择，并且您将只看到在您的时间窗内发生的更改。

@sel_cdc_cv1.sql

Rem changes classified for specific product groups
SELECT p1.prod_id, p2.prod_category, p1.prod_min_price,
p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id
AND operation$='UN';
PROMPT and especially the Electronics' ones - 3 records only
SELECT p1.prod_id, p2.prod_category, p1.prod_min_price,
p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id
AND operation$='UN'
AND p2.prod_category='Electronics';

在上述示例中，您将连回源表 PRODUCTS。设置 CDC 环境时，请确保此操作不应是必须执行的；如果可能的话，更改表应可用作独立源来同步目标环境。连回源表可以减少磁盘上存储的数据，但它将为源站点带来额外的负载。无法在源系统以外的系统上完成连回。
5.

您现在可以在目标系统上使用更改。

@consume_cdc1.sql

Prompt ... you can now consume the changes
rem arbitrary example, where we only track ELECTRONICS changes.This shall demonstrate
rem the flexibility as well as the responsibility of the client site (consumer) to deal
rem appropriately with the changes
INSERT into my_price_change_electronics
SELECT p1.prod_id, p1.prod_min_price, p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id
AND p2.prod_category='Electronics' AND operation$='UN';

COMMIT;

SELECT prod_id, prod_min_price, prod_list_price,
to_char(when,'dd-mon-yyyy hh24:mi:ss')
FROM my_price_change_electronics;

返回主题

5.激活预订并扩展预订窗口

到目前为止，您已经完成了初次使用同步 CDC 框架的练习。下面，您将了解如何在将要发生其他更改并且所有预定方均使用旧更改的环境下进行处理。

1.

对源表 PRODUCTS 执行其他 DML 操作。

@dml_cdc2.sql

PROMPT other changes will happen
UPDATE products
SET prod_min_price=prod_min_price*1.1
WHERE prod_min_price < 10;
COMMIT;
2. 同步 CDC 框架在更改表中透明地跟踪这些 DML 操作。

@show_cdc_ct2.sql

SELECT count(*) FROM prod_price_ct ;
PROMPT and especially the ELECTRONICS' ones
SELECT COUNT(*) FROM prod_price_ct p1, products p2
WHERE p1.prod_id=p2.prod_id AND p2.prod_category='Electronics';

 

3.

运行以下脚本查看产品数据。

@sel_cdc_cv1.sql

Rem changes classified for specific product groups

SELECT p1.prod_id, p2.prod_category, p1.prod_min_price,
p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id
AND operation$='UN';
        
PROMPT and especially the Electronics' ones - 3 records only
SELECT p1.prod_id, p2.prod_category, p1.prod_min_price,
p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id
AND operation$='UN'
AND p2.prod_category='Electronics';
4.

要通知 CDC 框架您已经完成了一组更改，必须清除并扩展预订。

@purge_cdc_sub_window1.sql

PROMPT purge old data from the subscription window
Rem this will NOT delete any records from the change table.It just tells the CDC system
Rem that the changed data used so fasr is no longer needed

EXEC DBMS_CDC_SUBSCRIBE.PURGE_WINDOW -
(SUBSCRIPTION_name => 'my_subscription_no_1');
PROMPT still the same number of records in the change table
PROMPT REMEMBER THE NUMBER OF ROWS !!!
SELECT COUNT(*) FROM prod_price_ct;
PROMPT ... change view is empty
SELECT COUNT(*) FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id AND p2.prod_category='Electronics';
5.

要使新更改反映到更改视图中，必须扩展更改视图的时间窗。

@ext_cdc_sub_window1.sql

PROMPT let's get the new change
Rem 'do it again Sam'
Rem first extend the window you want to see
EXEC DBMS_CDC_SUBSCRIBE.EXTEND_WINDOW -
(SUBSCRIPTION_name =>'my_subscription_no_1');

Rem ... now you will see exactly the new changed data since the last consumption
SELECT COUNT(*) FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id AND p2.prod_category='Electronics';

新更改视图确切显示了尚未使用的更改。这与更改表中的内容不同。
6.

现在使用新更改。由于您所关注的增量式更改存储在更改表（一个“普通”的 Oracle 表）中，因此可以使用数据库支持的任何语言或 SQL 结构。

@consume_cdc2.sql

PROMPT the new changes will be used
MERGE INTO my_price_change_Electronics t
USING (
SELECT p1.prod_id, p1.prod_min_price, p1.prod_list_price, commit_timestamp$
FROM my_prod_price_change_view p1, products p2
WHERE p1.prod_id=p2.prod_id AND p2.prod_category='Electronics' AND
operation$='UN') cv
ON (cv.prod_id = t.prod_id)
WHEN MATCHED THEN
UPDATE SET t.prod_min_price=cv.prod_min_price,
t.prod_list_price=cv.prod_list_price
WHEN NOT MATCHED THEN
INSERT VALUES (cv.prod_id, cv.prod_min_price, cv.prod_list_price,
commit_timestamp$);
COMMIT;
rem look at them
SELECT prod_id, prod_min_price, prod_list_price,
to_char(when,'dd-mon-yyyy hh24:mi:ss')
FROM my_price_change_electronics;

返回主题

6.运行发布方

发布方负责维护 CDC 框架，并确保在所有预订方已经使用了一组特定的更改信息后定期清除更改表。

1.

在登录 SH 模式的 SQL*Plus 会话中，运行 purge_ct.sql，或将下列 SQL 语句复制到 SQL*Plus 会话中：

@purge_cdc_ct.sql

PROMPT what to do to avoid overflow of change table ?PURGE IT !


Rem only the rows where we have no potential subscribers will be purged !!!
Rem we guarantee to keep all rows as long as we have subscribers which haven't
rem purged their window ...
exec DBMS_CDC_PUBLISH.purge_change_table('sh','prod_price_ct')
PROMPT REMEMBER THE NUMBER OF ROWS !!!
Rem you will have 18 entries for the last updated 9 records
SELECT COUNT(*) FROM prod_price_ct;
PROMPT this is exactly twice the number of changes we made with the second DML operation
SELECT COUNT(*) FROM prod_price_ct p1, products p2
WHERE p1.prod_id=p2.prod_id AND p2.prod_category='Electronics';

尽管没有为更改表的清除操作指定任何特定的时间窗，但 CDC 框架确保只清除预订者不再需要的那些更改。

返回主题

7.删除已使用的更改视图并清除预订窗口

1.

由于您已经使用了第二组更改，因此可以删除已使用的更改视图并清除预订窗口。您还将清除更改表。

@purge_cdc_sub_window2.sql

PROMPT ... purge the newly consumed data from the subscription window (2nd DML operation))
EXEC DBMS_CDC_SUBSCRIBE.PURGE_WINDOW -
(SUBSCRIPTION_name =>'my_subscription_no_1');
PROMPT purge all change tables again
EXEC DBMS_CDC_PUBLISH.purge_change_table('sh','prod_price_ct')
PROMPT ... and you will see an empty change
SELECT * FROM prod_price_ct;


更改表现在为空。因为已使用了所有更改，所以可以将其清除。

 

返回主题

8.清理 CDC 环境

清理 CDC 环境。为此，执行下列步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，运行 cleanup_cdc.sql，或将下列 SQL 语句复制到 SQL*Plus 会话中：

@cleanup_cdc.sql

PROMPT CLEAN UP
exec DBMS_CDC_SUBSCRIBE.drop_subscription -
(subscription_name=> 'my_subscription_no_1');
exec DBMS_CDC_PUBLISH.DROP_CHANGE_TABLE (OWNER => 'sh', -
CHANGE_TABLE_NAME => 'prod_price_CT', -
FORCE_FLAG => 'Y');
DROP TABLE my_price_change_electronics;
UPDATE products SET prod_list_price=prod_list_price/1.1
WHERE prod_min_price > 100;
UPDATE products SET prod_list_price=prod_list_price/1.1
WHERE prod_min_price < 10;
COMMIT;

返回主题

返回主题列表

从数据仓库传送到数据集市

除了中央数据仓库以外，MyCompany 还运行着多个小型部门数据集市。例如，产品部门需要接收按其主要产品类别进行划分的所有事务 SALES 数据以进行营销活动分析；只有 2000 年的 SALES 数据是相关的。您将使用 Oracle 的可传输表空间功能和 Oracle 的 LIST 分区解决业务问题。此外，为确保成功完成这个新表的生成过程，您将在 RESUMABLE 模式下运行语句，从而确保任何空间问题均不会导致创建过程失败。

注意： 为演示 resumable 语句的好处，您需要两个会话，以及相应的两个窗口。练习前请仔细阅读以下部分。

要从数据仓库传送到数据集市，执行下列步骤：

1. 启用 RESUMABLE 会话。
2. 创建一个新的表空间作为潜在的可传输表空间。
3. 在新表空间中创建一个 LIST 分区表。
4. 利用新的 RESUMABLE 语句功能实现高效的错误检测和处理。
5. 创建一个新的 RANGE_LIST 分区表（这是 Oracle9i 第 2 版的功能）。
6. 为可传输表空间准备元数据导出。

返回主题列表

1.启用 RESUMABLE 会话

使用 resumable 语句，Oracle 能够在发生可修复故障的情况下暂停并恢复大型数据库操作的执行。当前，语句可恢复的错误类型是空间限制和空间不足错误。当操作暂停时，您可以采取更正步骤解决错误情况。或者，可以注册过程来实现自动化的错误更正。解决错误情况后，暂停的语句将自动恢复并继续运行。如果在可选的指定暂停时间限制内未更正错误，该语句最终将失败。在本练习中，您将选择手动更正错误。

要了解如何完成此操作，执行以下步骤：

1.

该 SQL 语句使用两个相互嵌套的表函数。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@set_resumable.sql

PROMPT CLEAN UP
ALTER SESSION ENABLE RESUMABLE TIMEOUT 1200 NAME 
'create list partitioning';
这将使会话处于所谓的可恢复模式，对它进行命名并启用一个最大值为 1200 秒的暂停时间。


返回主题

2.创建一个新的表空间作为潜在可传输的表空间

现在，创建一个用于存储新 LIST 分区事实表的附加表空间。为此，执行以下步骤：

1.

首先，确保潜在的表空间 TRANSFER 不存在。在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@drop_TS.sql

DROP TABLESPACE my_obe_transfer INCLUDING CONTENTS AND DATAFILES;

该表空间已被删除。如果收到 ORA-959 消息，则可以将它忽略；这意味着该表空间不存在。
2. 在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

注意：如果对数据文件使用不同的驱动器或路径，则将需要对该脚本进行相应的编辑。

@create_ts.sql

CREATE TABLESPACE my_obe_transfer DATAFILE '/tmp/tt' SIZE 2M REUSE autoextend off;

故意将表空间定义得很小，以便目标表在创建表的过程中强制产生一个空间错误。

 

返回主题

 

3.在新表空间中创建一个 LIST 分区表

在窗口 1 中，您使用 CREATE TABLE AS SELEC 命令在新创建的表空间传输中创建了一个 LIST 分区表。

1.

在登录 SH 模式的 SQL*Plus 会话中，运行 create_list_part_table.sql。

注意：该语句将由于 TRANSFER 表空间中的空间问题而挂起并显示在警报日志中。当它挂起时，转到下一步。

@create_list_part_table.sql

DROP TABLE sales_prod_dept;
PROMPT create table in new TS that is too small
CREATE TABLE sales_prod_dept
(prod_category, prod_subcategory,cust_id, 
time_id,channel_id,promo_id, quantity_sold, amount_sold
) NOLOGGING TABLESPACE my_obe_transfer
PARTITION BY LIST (prod_category)
(PARTITION electronic_sales values ('Electronics'),
PARTITION hardware_sales values ('Hardware'),
PARTITION sw_other_sales values ('Software/Other'),
PARTITION p_and_a values ('Peripherals and Accessories'),
PARTITION photo_sales values ('Photo')
)
AS
SELECT p.prod_category, p.prod_subcategory, s.cust_id, s.time_id,s.channel_id,
s.promo_id, SUM(s.amount_sold) amount_sold, SUM(s.quantity_sold) quantity_sold
FROM sales s, products p, times t
WHERE p.prod_id=s.prod_id
AND s.time_id = t.time_id
AND t.fiscal_year=2000
GROUP BY prod_category, prod_subcategory,cust_id, s.time_id,channel_id, promo_id
;

返回主题

4.利用新的 RESUMABLE 语句功能实现高效的错误检测和处理

当 CREATE TABLE AS SELECT 在窗口 1 中运行时，可以在窗口 2 中控制在 RESUMABLE 模式下运行的所有会话的状态。

1.

在登录 SH 模式的 SQL*Plus 会话中，执行下列 SQL 语句：

@run_sel_w2.sql

SELECT NAME, STATUS, ERROR_MSG FROM dba_resumable;

只要不发生错误，该会话便保持正常的运行状态。一旦发生错误，会话状态将更改，并显示导致该问题的 ORA 错误。直到修复了问题，或直到达到超时限制时，RESUMABLE 会话中的当前语句才将暂停。错误还将显示在正在运行的实例的 alert.log 文件中。
2. 现在，手动修复问题。执行下列命令的单独表空间/数据文件定义。在登录 SYSTEM 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@fix_ts.sql

ALTER DATABASE DATAFILE '/tmp/tt' AUTOEXTEND ON NEXT 5M;

表空间现在可以由数据库本身自动扩展而不需要任何进一步的交互。

 

3.

修复错误后，暂停的会话将自动恢复并继续处理。检查数据词典视图中是否存在状态更改。在登录 SH 模式的 SQL*Plus 会话中，再次检查状态：

@run_sel_w2.sql

SELECT NAME, STATUS, ERROR_MSG FROM dba_resumable;

在前面挂起的窗口中恢复语句。

返回主题

5.在传输表空间中创建 Range-List 分区表

利用另一个分区策略 RANGE-LIST，您可以使用复合式分区方法基于两个逻辑属性对分区表进行细分。主要的分区策略是著名且最常用的范围分区策略。几乎在每个涉及滚动窗口操作的数据仓库实施中都会找到该实施。此外，每个范围分区均基于列表分区策略进行划分，从而实现一个更细的分区粒度来满足业务需要。

一个典型的利用 Range-List 分区的示例是全球零售环境，由于该环境基于时间范围分区（滚动窗口）和面向区域的列表分区，因此可以相互独立地维护特定区域的每个时间窗。

列表分区表使您可以将不同的无关值组划分到一个分区中。因此，现有分区定义未涉及的所有值均将引发 Oracle 错误消息。尽管这实施了业务约束（您不会在数据仓库中看到一个不存在的区域），但您无法始终确保不出现这样的违规情况。因此，Oracle 引入了创建默认列表分区（一种用于所有未定义的分区键值的全部捕获分区）这一功能。

Oracle 还引入了所谓的子分区模板技术，它是范围散列和范围列表复合式分区的一个常见可选元素。使用该模板，您可以轻松、方便地为每个表分区定义默认的子分区。Oracle 将在任何未显式定义子分区的分区中创建这些默认的子分区。该子句用于创建对称分区。

在本部分中，您将创建一个利用上述所有功能的范围列表分区表。

1.

在登录 SH 模式的 SQL*Plus 会话中，运行下列 SQL 语句：

@create_new_range_list.sql

CREATE TABLE sales_rlp
COMPRESS
TABLESPACE MY_OBE_TRANSFER
PARTITION BY RANGE (time_id)
SUBPARTITION BY LIST (channel_id)
SUBPARTITION TEMPLATE
( SUBPARTITION direct values (3),
SUBPARTITION internet values (4),
SUBPARTITION partner values (2),
SUBPARTITION other values (DEFAULT)
)
(PARTITION SALES_before_1999 VALUES LESS THAN (TO_DATE('01-JAN-1999','DD-MON-YYYY')),
PARTITION SALES_Q1_1999 VALUES LESS THAN (TO_DATE('01-APR-1999','DD-MON-YYYY')),
PARTITION SALES_Q2_1999 VALUES LESS THAN (TO_DATE('01-JUL-1999','DD-MON-YYYY')),
PARTITION SALES_Q3_1999 VALUES LESS THAN (TO_DATE('01-OCT-1999','DD-MON-YYYY')),
PARTITION SALES_Q4_1999 VALUES LESS THAN (TO_DATE('01-JAN-2000','DD-MON-YYYY')),
PARTITION SALES_Q1_2000 VALUES LESS THAN (TO_DATE('01-APR-2000','DD-MON-YYYY')),
PARTITION SALES_Q2_2000 VALUES LESS THAN (TO_DATE('01-JUL-2000','DD-MON-YYYY')),
PARTITION SALES_Q3_2000 VALUES LESS THAN (TO_DATE('01-OCT-2000','DD-MON-YYYY')),
PARTITION SALES_Q4_2000 VALUES LESS THAN (MAXVALUE) NOCOMPRESS)
AS
SELECT * FROM sales sample(10);

该示例还显示了如何在创建初始表的过程中轻松地创建压缩和未压缩的分区。分区和模板名称由所有子分区继承，从而使子分区的命名和标识变得更简单、更方便。

 
2. 在登录 SH 模式的 SQL*Plus 会话中，运行下列 SQL 语句：

@show_range_list_names.sql

PROMPT as you can see, the partition and template name is inherited to all the subpartitions,
PROMPT thus making the naming - and identification - of subpartitions much easier ...
select partition_name, subpartition_name, high_value from user_tab_subpartitions
where table_name='SALES_RLP';

列表和范围列表分区表将不同的无关值组划分到一个分区中。因此，现有分区定义未涉及的所有值均将引发 Oracle 错误消息。我们现在将为前面创建的列表分区表 sales_prod_dept 创建一个默认的列表分区（一种用于所有未定义的分区键值的全部捕获分区）。

 
3.

在登录 SH 模式的 SQL*Plus 会话中，运行下列 SQL 语句：

@cr_default_list_part.sql 

PROMPT add a new partition to sales_prod_dept that does NOT have a DEFAULT partition
Rem it is added just like for a range partitioned table
ALTER TABLE sales_prod_dept ADD PARTITION gameboy_sales VALUES ('Gameboy');

PROMPT now add another one, covering the DEFAULT value
ALTER TABLE sales_prod_dept ADD PARTITION all_other_sales VALUES (DEFAULT);

PROMPT view the data dictionary
select partition_name, high_value 
from   user_tab_partitions 
where table_name='SALES_PROD_DEPT';

所有具有未定义的分区键的记录都将存储在这个新分区 other_sales 中。但采用默认分区将改变向该表中添加新分区的方法。 Conceptually, you cover all possible values for the partitioning key—the defined ones and "all others"—as soon as you have created a default partition.

要向该表中添加新分区，必须在逻辑上将默认分区拆分为一个具有一组定义键和一个新的默认分区的新分区，但仍涵盖“所有其他值”（现在减少了为新分区指定的键）。添加新分区的任何尝试（类似前面进行的尝试）都将引发 ORA 错误消息。
4.

在登录 SH 模式的 SQL*Plus 会话中，运行下列 SQL 语句：

@split_default_list_part.sql

PROMPT Unlike the first time, we cannot simply add a new partition
PROMPT raises ORA-14323:cannot add partition when DEFAULT partition exists
Rem we cannot be sure whether the new value already exists in the DEFAULT partition
ALTER TABLE sales_prod_dept ADD PARTITION undefined_sales VALUES ('Undefined');

这将引发一个 Oracle 错误。运行下列命令或脚本 split_default_list_part_b.sql

@split_default_list_part_b.sql

PROMPT so we have to SPLIT the default partition to ensure that 
PROMPT all potential values of 'Undefined' are in the new partition
       
ALTER TABLE sales_prod_dept SPLIT 
PARTITION other_sales VALUES ('Undefined') INTO 
(PARTITION undefined_sales, PARTITION other_sales);
PROMPT control the data dictionary
PROMPT Note that without specifying any tablespace, the default 
PROMPT tablespace of the partitioned table is used
select partition_name, tablespace_name, high_value 
from user_tab_partitions 
where table_name='SALES_PROD_DEPT';

返回主题

6.为可传输的表空间准备元数据导出

要为传输准备表空间，只需导出表空间的元数据信息。该导出以及表空间的物理副本可用于将表空间导入到另一个数据库实例中。在 Oracle9i 之前，必须在同一操作系统上运行相同数据库块大小的目标数据库。Oracle9i 解除了同一块大小的限制。Oracle 数据库 10g 通过引入异构可传输表空间进一步解除了此限制。

要演示该功能，执行下列步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行以下 SQL 脚本：

@make_ts_ro.sql

ALTER TABLESPACE my_obe_transfer READ ONLY;

这确保表空间中的数据不会发生进一步的更改。
2. 您现在可以按如下所示导出表空间 my_obe_ transfer 的数据词典信息：

@export_metadata.sql

或者，您可以创建一个用于存储导出转储文件的特定目录：

CREATE DIRECTORY my_obe_dump_dir as '/home/oracle/wkdir/';

现在，您只能导出元数据。请比较文件大小与表空间大小，以体会提取所有数据而不仅仅提取元数据的含义。

expdp \'/ as sysdba\' DIRECTORY=my_obe_dump_dir DUMPFILE= meta_MY_OBE_TRANSFER.dmp

注意：该脚本可能需要几分种的时间才能完成运行。

注意：可以通过删除 SQL*Plus“host”命令在 DOS 提示符下使用同一导出语法。

 

3.

将表空间信息导入到另一个数据库之前，您将通过执行以下 SQL 脚本清理并重置会话状态：

@cleanup_tts.sql 

DROP TABLESPACE transfer INCLUDING CONTENTS AND DATAFILES;
ALTER SESSION DISABLE RESUMABLE;
4.

尝试访问表 sales_prod_dept 将引发错误。该表已不再存在。

@sel_spd.sql

SELECT count(*) 
from sales_prod_dept;

 

返回主题

返回主题列表

清理

要清理您的环境，您需要执行以下步骤：

1.

在登录 SH 模式的 SQL*Plus 会话中，执行下列命令：

SET SERVEROUTPUT ON
EXEC dw_handsOn.cleanup_modules

返回主题列表

总结

<包含对用户通过执行所有步骤完成的任务的简短摘要。您还可以包含您认为相关并且有帮助的其他信息、可选的其他练习（帮助用户了解有关相关功能的更多信息）以及相关的主题链接（如下面的“相关信息”部分所述）。>

在本课程中，您学习了如何执行下列任务：

执行多表插入
执行 Upsert
使用表函数
使用同步 CDC 捕获并使用增量式源更改
从数据仓库传送到数据集市

返回主题列表

将光标置于该图标上可以隐藏所有屏幕截图。

 

 

E-mail this page
打印机视图 打印机视图