SQL> CREATE OR REPLACE DIRECTORY source_dir as 'C:\oracle\oradata\source'; --源文件目录 Directory created. SQL> grant connect, dba to stenny identified by stenny; Grant succeeded. SQL> CREATE OR REPLACE DIRECTORY source_dir as 'C:\oracle\oradata\source'; Directory created. SQL> CREATE OR REPLACE DIRECTORY target_dir as 'C:\oracle\oradata\target'; Directory created. SQL> CREATE OR REPLACE DIRECTORY log_dir as 'C:\oracle\oradata\log'; Directory created. SQL> grant read on directory source_dir to stenny; Grant succeeded. SQL> grant write on directory target_dir to stenny; Grant succeeded. SQL> grant write on directory log_dir to stenny; Grant succeeded.  
	

    数据抽取— ORACLE_LOADER 
    平文件的数据源一般有两种格式：定长格式或者分隔符格式，ORACLE_LOADER对这两种格式都提供支持。如果是分隔符格式的源文件，使用关键字FIELDS TERMINATED BY指定分隔符；如果是定长格式文件，使用关键字POSITION指定数据列的位置。 
    --数据源文件product1.dat 
    1,Bicycle,JiangSu 
    2,Camps,ZheJiang 
    3,Wearings,SiChuan 
    4,Gloves,SiChuan 
    5,Food,YunNan 
    6,Shoes,NULL 
    使用EXTERNAL ORGANIZATIONAL关键字构建外部表，注意DRIVER关键字指定了ORACLE_LOADER驱动；DEFAULT DIRECTORY子句指定了默认的操作目录；在ACCESS PARAMETERS子句中定义了分隔符、坏文件、LOG文件；LOCATION关键字指定了外部表依存的数据文件，如果存在多个文件，使用逗号分隔。 

	
	

		SQL> ed Wrote file afiedt.buf  1 CREATE TABLE stenny_ext_product 2 (product_id NUMBER(4), 3 product_name VARCHAR2(20), 4 location VARCHAR2(25) 5 ) 6 ORGANIZATION EXTERNAL 7 ( 8 TYPE ORACLE_LOADER 9 DEFAULT DIRECTORY source_dir 10 ACCESS PARAMETERS 11 ( 12 records delimited by newline 13 badfile log_dir:'bad_product.dat' 14 logfile log_dir:'product.log' 15 fields terminated by ',' 16 missing field values are null 17 ( product_id, product_name, location ) 18 ) 19 LOCATION ('product1.dat') 20 ) 21* REJECT LIMIT UNLIMITED SQL> /  Table created.  SQL> select * from stennY_ext_product;  PRODUCT_ID PRODUCT_NAME LOCATION ---------- -------------------- ------------------------- 1 Bicycle JiangSu 2 Camps ZheJiang 3 Wearings SiChuan 4 Gloves SiChuan 5 Food YunNan 6 Shoes NULL  6 rows selected.  
	

    经过上面的步骤，我们将数据库之外的一个平文件通过访问驱动程序ORACLE_LOADER与一个数据库表STENNY_EXT_PRODUCT建立了映射关系。我们对这个外部表可以进行排序，表连接等只读操作。

 

    数据清洗与数据整合 
    在典型的数据仓库系统中，事实表中的大部分字段都采用KEY的形式进行存储。在我们的例子中，我们将对LOCATION列进行LOOKUP，满足地区标准的数据被视为正确数据并附以标准键值；如果出现在标准表中不存在的纪录，我们将认为其为脏数据。脏数据将被另行处理。我们的标准表为LOC_STD，正确记录的Staging表为STG_PRODUCT，异常记录的Staging表为STG_EXCEP。建表语句与测试数据见附录A。在数据文件中，ID为6的产品没有对应的Location，该记录将被视为异常记录。通过一个简单的存储过程，我们将STENNY_EXT_PRODUCT与LOC_STD表进行连接，正确的记录与异常记录将被写入对应的Staging表中。 

	
	

		--proc_txn_product CREATE OR REPLACE PROCEDURE proc_txn_product AS BEGIN insert into stg_product select product_id,product_name,loc_id from stenny_ext_product,loc_std where loc_std.loc_name=stenny_ext_product.location; insert into stg_excep select * from stenny_ext_product where product_id not in (select product_ id from stg_product); commit; END proc_txn_product  
	

    执行存储过程可以得到如下的结果： 

	
	

		SQL> exec proc_txn_product; PL/SQL procedure successfully completed. SQL> select * from stg_product; PRODUCT_ID PRODUCT_NAME LOC_ID ---------- -------------------- ---------- 1 Bicycle 1 2 Camps 2 3 Wearings 3 4 Gloves 3 5 Food 4  SQL> select * from stg_excep;  PRODUCT_ID PRODUCT_NAME LOCATION ---------- -------------------- ------------------------- 6 Shoes NULLEND proc_txn_product  
	

    数据分发 – ORACLE_DATAPUMP 
    在经过了上面的抽取、清洗整合的步骤后，我们将使用ORACLE_DATAPUMP将staging表中的数据写入目标文件中。典型的数据仓库系统最终的目标是数据库，而本文中的ECCD流程可以是数据准备的一个过程，众所周知，数据仓库的刷新频率与Staging区域可能不同，目标文件可以被传输到数据仓库系统，并按照其刷新频率进行第二步加载。顺便提一句，DBMS_SCHEDULER可以用来完成在数据库之间进行文件传输。 

    下面的代码演示了如何使用ORACLE_DATAPUMP驱动访问程序unload数据库表的过程。 

	
	

		SQL> ed Wrote file afiedt.buf  1 CREATE TABLE tgt_product 2 ORGANIZATION EXTERNAL (TYPE ORACLE_DATAPUMP 3 DEFAULT DIRECTORY target_dir 4 LOCATION ('tgt_product.dmp')) 5 PARALLEL 2 6 AS 7 SELECT product_id, 8 product_name, 9 loc_id 10* FROM stg_product SQL> /  Table created.  SQL> select * from tgt_product;  PRODUCT_ID PRODUCT_NAME LOC_ID ---------- -------------------- ---------- 1 Bicycle 1 2 Camps 2 3 Wearings 3 4 Gloves 3 5 Food 4  
	

    同时，在target_dir目录下可以看到创建的TGT_PRODUCT.DMP文件，该文件可以被加载到真正的目标数据库，但是必须使用ORACLE_DATAPUMP访问驱动程序。

   定时调度 – DBMS_SCHEDULER 
    首先我们将使用proc_file_watcher存储过程每5分钟对某一个目录进行扫描等待文件。如果文件存在，调用proc_txn_product对文件进行处理，否则睡眠5分钟。这个文件扫描过程在周二早上4点钟被激活，如果超过8点文件还是没有到，可以抛出用户异常或者调用UTL_SMTP包发邮件通知相关责任人。 

	
	

		create or replace procedure proc_file_watcher is v_exists boolean; v_file_length number; v_blocksize number; begin <<L_sleeping_child>> if to_char(sysdate,'hh24')>='08' then --超时，可以调用UTL_SMTP null; else utl_file.fgetattr('SOURCE_DIR','product1.dat',v_exists,v_file_length,v_blocksize); if v_exists then dbms_output.put_line('File there!'); proc_txn_product; else dbms_output.put_line('404 Error'); dbms_lock.sleep(300); goto L_sleeping_child; end if; end if; end proc_file_watcher;  
	

    现在，我们只需要将这个文件扫描程序加入调度，每个周二早上4点钟开始运行即可。 
    --创建程序 

	
	

		SQL> ed Wrote file afiedt.buf  1 BEGIN 2 DBMS_SCHEDULER.CREATE_PROGRAM( 3 program_name=>'STENNY.STP_PROC_FILE_WATCHER', 4 program_action=>'STENNY.PROC_FILE_WATCHER', 5 program_type=>'STORED_PROCEDURE', 6 comments=>'Firing the ETL process if file arrives', 7 enabled=>TRUE); 8* END; SQL> /  PL/SQL procedure successfully completed. --创建调度 SQL> ed Wrote file afiedt.buf  1 BEGIN 2 SYS.DBMS_SCHEDULER.CREATE_SCHEDULE( 3 repeat_interval => 'FREQ=WEEKLY;BYDAY=TUE;BYHOUR=8;BYMINUTE=0;BYSECOND=0', 4 start_date => to_timestamp_tz('2004-04-27 US/Central', 'YYYY-MM-DD TZR'), 5 comments => 'Tuesday AM Schedule', 6 schedule_name => '"STENNY"."SCS_TXN_PROD"'); 7* END; SQL> /  PL/SQL procedure successfully completed. 创建工作 SQL> ed Wrote file afiedt.buf  1 BEGIN 2 SYS.DBMS_SCHEDULER.CREATE_JOB( 3 job_name => 'STENNY.SCJ_TXN_PROD', 4 program_name => 'STENNY.STP_PROC_FILE_WATCHER', 5 schedule_name => 'STENNY.SCS_TXN_PROD', 6 comments => 'Start the ETL process on Tuesday', 7 auto_drop => FALSE, 8 enabled => TRUE); 9* END; SQL> /  PL/SQL procedure successfully completed. --进行测试 SQL> select count(*) from stenny.stg_product;  COUNT(*) ---------- 0 SQL> EXEC DBMS_SCHEDULER.RUN_JOB('STENNY.SCJ_TXN_PROD',FALSE);  PL/SQL procedure successfully completed.  SQL> select count(*) from stenny.stg_product;  COUNT(*) ----------  
	

    附录A, 创建测试表的语句 

--stg_excep Create table stg_excep as select * from stenny_ext_product where 1=2; --stg_product CREATE TABLE STG_PRODUCT ( PRODUCT_ID NUMBER, PRODUCT_NAME VARCHAR2(20), LOC_ID NUMBER ); --loc_std CREATE TABLE LOC_STD ( LOC_ID NUMBER, LOC_NAME VARCHAR2(20) ); INSERT INTO LOC_STD ( LOC_ID, LOC_NAME ) VALUES ( 1, 'JiangSu'); INSERT INTO LOC_STD ( LOC_ID, LOC_NAME ) VALUES ( 2, 'ZheJiang'); INSERT INTO LOC_STD ( LOC_ID, LOC_NAME ) VALUES ( 3, 'SiChuan'); INSERT INTO LOC_STD ( LOC_ID, LOC_NAME ) VALUES ( 4, 'YunNan');