datastage在业务逻辑复杂的情况下如何应用

初级会员 注册时间: 2008-06-02 19:32:03 文章: 14 离线

最近两周才开始学习datastage，基本的内容了解的差不多，但是想想如何使用datastage来实现
刚刚完成的复杂业务数据抽取，觉得有点没有头绪。希望大家给出一些好的想法。
举个例子吧，我们的数据转换中的一个procedure，这个procedure在datastage中该如何实现。
我知道datastage中也有procedure stage，但不知道如何实现。
简单的说明一下这个procedure：
主要是对一个游标进行循环，
然后经过中间的复杂的逻辑转换
将每一个循环的结果插入到表中。
因为刚接触datastage，所以如何用datastage来实现，还真是搞不来。麻烦大家帮忙，谢谢。
PROCEDURE GSCS_BY_RAW_MATERIAL (p_run_date_ NUMBER, p_site VARCHAR2)
AS
i_site VARCHAR2 (10);
i_loacal_currency VARCHAR (6);
i_itemno VARCHAR2 (30);
i_idesc VARCHAR2 (60);
i_itemtype VARCHAR2 (15);
i_facility VARCHAR2 (15);
i_soqty NUMBER (22, 5);
i_so_usd_amt NUMBER (22, 5);
i_groupid NUMBER (22, 0);
i_sotype VARCHAR2 (4);
i_user VARCHAR2 (20);
i_errorcode NUMBER;
i_errortext VARCHAR2 (200);
i_datetime DATE;
i_date_ NUMBER (10);
i_soamt NUMBER (22, 5);
i_count NUMBER;
i_isconsiged VARCHAR2 (30);
i_itemcatalid VARCHAR2 (10);
i_keypartset VARCHAR2 (30);
i_nonkey9b NUMBER (22, 0);
i_nonkeynon9bconsign NUMBER (22, 0);
i_nonkeynon9bnonconsign NUMBER (22, 0);
i_tableexist NUMBER;
i_sql VARCHAR2(2000);
--end added

--************************************************************
CURSOR cur IS
SELECT a.facility, a.itemno,
(a.openbalance - issueqty + adjustqty + receiveqty
) AS soqty,b.itemdesc, b.itemtype
FROM ginv_ili a,ginv_iim b,
(SELECT DISTINCT b.site, b.facility, b.warehouse,
b.LOCATION
FROM globallocation b,
invgroup d
WHERE ( UPPER (d.grouplevel2) =
UPPER ('2.Material-Non Key parts')
OR UPPER (d.grouplevel2) =
UPPER ('1.Material-Key parts')
)
AND UPPER (d.locationgroup) =
UPPER (TRIM (b.locationgroup))
AND b.site = p_site) c
WHERE a.site = c.site
AND a.facility = c.facility
AND a.warehouse = c.warehouse
AND a.LOCATION = c.LOCATION
and a.site = b.site
and a.facility = b.facility
and a.itemno = b.itemno
union all
--Add IQC
SELECT a.FACILITY,a.ITEMNO,a.INSPECTIONQTY AS soqty,b.itemdesc, b.itemtype
FROM ginv_IQC a,ginv_iim b
where a.site = p_site
and a.site = b.site
and a.facility = b.facility
and a.itemno = b.itemno
and b.itemtype in ('7','9','R');

c_row_cur cur%ROWTYPE;
BEGIN
SELECT username
INTO i_user
FROM v$session
WHERE SID = (SELECT DISTINCT SID
FROM v$mystat);

SELECT SYSDATE, sysdate_
INTO i_datetime, i_date_
FROM DUAL;

SELECT querycondition
INTO i_loacal_currency
FROM globalsysparameter
WHERE UPPER (TRIM (key1)) = 'CURRENCY' AND UPPER (TRIM (key2)) = p_site;

SELECT sequenceid
INTO i_nonkey9b
FROM invgroup
WHERE UPPER (grouplevel1) = UPPER ('Prod. INV')
AND UPPER (grouplevel4) = UPPER ('Non Key Parts-9B Parts');

SELECT sequenceid
INTO i_nonkeynon9bconsign
FROM invgroup
WHERE UPPER (grouplevel1) = UPPER ('Prod. INV')
AND UPPER (grouplevel4) = UPPER ('Non Key Parts-Non 9B Parts')
AND UPPER (grouplevel5) = UPPER ('Consign');

SELECT sequenceid
INTO i_nonkeynon9bnonconsign
FROM invgroup
WHERE UPPER (grouplevel1) = UPPER ('Prod. INV')
AND UPPER (grouplevel4) = UPPER ('Non Key Parts-Non 9B Parts')
AND UPPER (grouplevel5) = UPPER ('Non-Consign');

--create global temporary table ginv_cmf
i_tableexist :=0;
SELECT count(1) INTO i_tableexist
FROM USER_TABLES
WHERE TABLE_NAME = 'GLOBAL_TEMPORARY_GINV_CMF';

IF i_tableexist <> 1 THEN
i_sql := 'CREATE GLOBAL TEMPORARY TABLE GLOBAL_TEMPORARY_GINV_CMF
ON COMMIT PRESERVE ROWS
AS
SELECT site,facility,itemno,orgstdcost
FROM ginv_cmf
WHERE site = ''' || p_site || '''';
EXECUTE IMMEDIATE i_sql;
ELSE
i_sql := 'DELETE FROM GLOBAL_TEMPORARY_GINV_CMF';
EXECUTE IMMEDIATE i_sql;
i_sql := 'INSERT INTO GLOBAL_TEMPORARY_GINV_CMF
SELECT site,facility,itemno,orgstdcost
FROM ginv_cmf
WHERE site = ''' || p_site || '''';
EXECUTE IMMEDIATE i_sql;
COMMIT;
END IF;
--create global temporary table gcost_hqt
i_tableexist := 0;
SELECT count(1) INTO i_tableexist
FROM USER_TABLES
WHERE TABLE_NAME = 'GLOBAL_TEMPORARY_GCOST_HQT';

IF i_tableexist <> 1 THEN
i_sql :='CREATE GLOBAL TEMPORARY TABLE GLOBAL_TEMPORARY_GCOST_HQT
ON COMMIT PRESERVE ROWS
AS
SELECT hqsite,hqfac,hqprod,case when upper(cap)=''Y'' then ''Consign''
when upper(cap)=''N'' then ''Non-Consign''
else ''Non-Consign''
end as cap,hqeff,hqdis
FROM gcost_hqt
WHERE UPPER (TRIM (hqsite)) ='|| 'UPPER (''' || p_site || ''')'||
' AND hqeff <= ' || i_date_ ||
' AND hqdis >= ' || i_date_;
EXECUTE IMMEDIATE i_sql;
ELSE
i_sql := 'DELETE FROM GLOBAL_TEMPORARY_GCOST_HQT';
EXECUTE IMMEDIATE i_sql;
i_sql := 'INSERT INTO GLOBAL_TEMPORARY_GCOST_HQT
SELECT hqsite,hqfac,hqprod,case when upper(cap)=''Y'' then ''Consign''
when upper(cap)=''N'' then ''Non-Consign''
else ''Non-Consign''
end as cap,hqeff,hqdis
FROM gcost_hqt
WHERE UPPER (TRIM (hqsite)) ='|| 'UPPER (''' || p_site || ''')'||
' AND hqeff <=' || i_date_ ||
' AND hqdis >=' || i_date_;
EXECUTE IMMEDIATE i_sql;
COMMIT;
END IF;

--create global temporary table gcost_mbmic
i_tableexist := 0;
SELECT count(1) INTO i_tableexist
FROM USER_TABLES
WHERE TABLE_NAME = 'GLOBAL_TEMPORARY_GCOST_MBMIC';

IF i_tableexist <> 1 THEN
i_sql :='CREATE GLOBAL TEMPORARY TABLE GLOBAL_TEMPORARY_GCOST_MBMIC
ON COMMIT PRESERVE ROWS
AS
select a.bmwhs,a.bchld,b.sequenceid,b.grouplevel5,b.grouplevel2
from gcost_mbmic a,invgroup b,gcost_itemcatalog c
WHERE a.bdsite = ''' || p_site ||''''||
' AND a.iskeyparts = ''' || 'Y' || '''' ||
' AND a.bdeff <= ' || i_date_ ||
' AND a.bddis >=' || i_date_ ||
' AND a.itemcatalid = c.icid AND UPPER (b.grouplevel4) = UPPER (c.ic2)' ||
' AND UPPER (b.grouplevel1) = UPPER (''Prod. INV'')';
EXECUTE IMMEDIATE i_sql;
ELSE
i_sql := 'DELETE FROM GLOBAL_TEMPORARY_GCOST_MBMIC';
EXECUTE IMMEDIATE i_sql;
i_sql := 'INSERT INTO GLOBAL_TEMPORARY_GCOST_MBMIC
select a.bmwhs,a.bchld,b.sequenceid,b.grouplevel5,b.grouplevel2
from gcost_mbmic a,invgroup b,gcost_itemcatalog c
WHERE a.bdsite = ''' || p_site ||''''||
' AND a.iskeyparts = ''' || 'Y' || '''' ||
' AND a.bdeff <= ' || i_date_ ||
' AND a.bddis >=' || i_date_ ||
' AND a.itemcatalid = c.icid AND UPPER (b.grouplevel4) = UPPER (c.ic2)' ||
' AND UPPER (b.grouplevel1) = UPPER (''Prod. INV'')';
EXECUTE IMMEDIATE i_sql;
COMMIT;
END IF;

--create global temporary table gcost_mdmaltmic
i_tableexist := 0;
SELECT count(1) INTO i_tableexist
FROM USER_TABLES
WHERE TABLE_NAME = 'GLOBAL_TEMP_GCOST_MDM';

IF i_tableexist <> 1 THEN
i_sql := 'CREATE GLOBAL TEMPORARY TABLE GLOBAL_TEMP_GCOST_MDM
ON COMMIT PRESERVE ROWS
AS
select a.bmwhs,a.brchld,b.sequenceid,b.grouplevel5,b.grouplevel2
from gcost_mdmaltmic a,invgroup b,gcost_itemcatalog c
WHERE a.bdsite = ''' || p_site ||''''||
' AND a.iskeyparts = ''' || 'Y' || '''' ||
' AND a.brdeff <= ' || i_date_ ||
' AND a.beddis >=' || i_date_ ||
' AND a.itemcatalid = c.icid AND UPPER (b.grouplevel4) = UPPER (c.ic2)' ||
' AND UPPER (b.grouplevel1) = UPPER (''Prod. INV'')';
EXECUTE IMMEDIATE i_sql;
ELSE

i_sql := 'DELETE FROM GLOBAL_TEMP_GCOST_MDM';
EXECUTE IMMEDIATE i_sql;

i_sql :='INSERT INTO GLOBAL_TEMP_GCOST_MDM
select a.bmwhs,a.brchld,b.sequenceid,b.grouplevel5,b.grouplevel2
from gcost_mdmaltmic a,invgroup b,gcost_itemcatalog c
WHERE a.bdsite = ''' || p_site ||''''||
' AND a.iskeyparts = ''' || 'Y' || '''' ||
' AND a.brdeff <= ' || i_date_ ||
' AND a.beddis >=' || i_date_ ||
' AND a.itemcatalid = c.icid AND UPPER (b.grouplevel4) = UPPER (c.ic2)' ||
' AND UPPER (b.grouplevel1) = UPPER (''Prod. INV'')';
EXECUTE IMMEDIATE i_sql;

COMMIT;

END IF;

--create global temporary table invgroup, keypartparameter
i_tableexist := 0;
SELECT count(1) INTO i_tableexist
FROM USER_TABLES
WHERE TABLE_NAME = 'GLOBAL_TEMPORARY_YEAR07_KP';

IF i_tableexist <> 1 THEN

i_sql := 'CREATE GLOBAL TEMPORARY TABLE GLOBAL_TEMPORARY_YEAR07_KP
ON COMMIT PRESERVE ROWS
AS
SELECT facility,keypartitem,keypartitemdesc,keypartset,grouplevel4,grouplevel2
FROM invgroup, keypartparameter
WHERE UPPER (grouplevel1) = UPPER (''Prod. INV'')
AND UPPER (grouplevel4) = UPPER (keypartset)';

EXECUTE IMMEDIATE i_sql;
ELSE

i_sql := 'DELETE FROM GLOBAL_TEMPORARY_YEAR07_KP';

EXECUTE IMMEDIATE i_sql;

i_sql := 'INSERT INTO GLOBAL_TEMPORARY_YEAR07_KP
SELECT facility,keypartitem,keypartitemdesc,keypartset,grouplevel4,grouplevel2
FROM invgroup, keypartparameter
WHERE UPPER (grouplevel1) = UPPER (''Prod. INV'')
AND UPPER (grouplevel4) = UPPER (keypartset)';

EXECUTE IMMEDIATE i_sql;
COMMIT;

END IF;

--*************************************
--OPEN CURSOR
--*************************************

begin
OPEN cur;
LOOP
FETCH cur INTO c_row_cur;

EXIT WHEN cur%NOTFOUND;

i_facility := c_row_cur.FACILITY;
i_itemno := c_row_cur.ITEMNO;
i_soqty := c_row_cur.soqty;
i_idesc := c_row_cur.itemdesc;
i_itemtype := c_row_cur.itemtype;

IF i_itemtype = '7' OR i_itemtype = '9' OR i_itemtype = 'R'
THEN
BEGIN
--**************************************
--second point modified by jerry g wang
--**************************************
SELECT i_soqty*orgstdcost
INTO i_soamt
FROM GLOBAL_TEMPORARY_GINV_CMF
WHERE (facility =i_facility OR facility =' ')
AND itemno = i_itemno
AND ROWNUM = 1;

EXCEPTION
WHEN NO_DATA_FOUND
THEN
i_soamt := 0;
END;

BEGIN
--**************************************
--third point modified by jerry g wang
--**************************************
SELECT cap
INTO i_isconsiged
FROM GLOBAL_TEMPORARY_GCOST_HQT
WHERE hqfac =i_facility
AND UPPER (TRIM (hqprod)) = UPPER (TRIM (i_itemno))
AND ROWNUM = 1;

EXCEPTION
WHEN NO_DATA_FOUND
THEN
i_isconsiged := 'Non-Consign';
END;

BEGIN
--GLOBAL_TEMPORARY_GCOST_MBMIC
SELECT sequenceid
INTO i_groupid
FROM GLOBAL_TEMPORARY_GCOST_MBMIC
WHERE bmwhs = i_facility
AND bchld = i_itemno
AND UPPER (grouplevel5) = UPPER (i_isconsiged)
AND ROWNUM = 1;

EXCEPTION
WHEN NO_DATA_FOUND
THEN
BEGIN
SELECT sequenceid
INTO i_groupid
FROM GLOBAL_TEMP_GCOST_MDM
WHERE bmwhs = i_facility
AND brchld = i_itemno
AND UPPER (grouplevel5) = UPPER (i_isconsiged)
AND ROWNUM = 1;

EXCEPTION
WHEN NO_DATA_FOUND
THEN
i_groupid := 0;
END;
END;

IF i_groupid = 0 then
BEGIN
--**************************************
--eight point modified by jerry g wang
--**************************************

SELECT GROUPLEVEL4
INTO i_keypartset
FROM GLOBAL_TEMPORARY_YEAR07_KP
WHERE facility = i_facility
AND ( keypartitem = SUBSTR (i_itemno , 1, 2)
OR keypartitem = SUBSTR (i_itemno, 1, 5)
)
AND UPPER (i_idesc) LIKE '%' || UPPER (keypartitemdesc) ||'%'
AND ROWNUM = 1;

SELECT sequenceid
INTO i_groupid
FROM invgroup
WHERE UPPER (grouplevel1) = UPPER ('Prod. INV')
AND UPPER (grouplevel4) = UPPER (i_keypartset)
AND UPPER (grouplevel5) = UPPER (i_isconsiged);

EXCEPTION
WHEN NO_DATA_FOUND
THEN
i_groupid := 0;
END;

END IF;

IF i_groupid = 0
THEN
IF UPPER (SUBSTR (i_itemno, 1, 2)) = '9B' THEN
i_groupid := i_nonkey9b;
ELSE
IF i_isconsiged = 'Y'
THEN
i_groupid := i_nonkeynon9bconsign;
ELSE
i_groupid := i_nonkeynon9bnonconsign;
END IF;
END IF;

IF i_soamt > 0
THEN
i_so_usd_amt := i_soamt * gscsgetexchgrate (p_site, i_loacal_currency, 'USD', p_run_date_);
ELSE
i_so_usd_amt := i_soqty
* gscsgetpacnprice (p_site, i_facility, i_itemno);
END IF;

INSERT INTO globalinvdata
(itemno, site, facility, inventoryqty,
inventoryamount, invgroupid, getdate,
lastmdate, lastmuser
)
VALUES (i_itemno, p_site, i_facility, i_soqty,
i_so_usd_amt, i_groupid, i_datetime,
p_run_date_, i_user
);

COMMIT;

ELSE

IF i_soamt > 0
THEN
i_so_usd_amt := i_soamt * gscsgetexchgrate (p_site, i_loacal_currency, 'USD', p_run_date_);
ELSE
i_so_usd_amt :=
i_soqty
* gscsgetpacnprice (p_site, i_facility, i_itemno);
END IF;

INSERT INTO globalinvdata
(itemno, site, facility, inventoryqty,
inventoryamount, invgroupid, getdate,
lastmdate, lastmuser
)
VALUES (i_itemno, p_site, i_facility, i_soqty,
i_so_usd_amt, i_groupid, i_datetime,
p_run_date_, i_user
);

COMMIT;
END IF;
END IF;

END LOOP;

CLOSE cur;

INSERT INTO qty_err_log_table
(sp, errcode,
errdescription, generatetime
)
VALUES (p_site || ' FROCEDURE:GSCS_BY_Raw_Material', 0,
'EXEC Successful!', i_datetime
);

COMMIT;
END;
EXCEPTION
WHEN OTHERS
THEN
ROLLBACK;
i_errorcode := SQLCODE;
i_errortext :=
SUBSTR ( '('
--|| i_site
|| p_site
|| ', '
|| i_facility
|| ', '
|| i_itemno
|| ')'
|| SQLERRM,
1,
200
);

INSERT INTO qty_err_log_table
(sp,
errcode, errdescription, generatetime
)
VALUES (p_site || ' FROCEDURE:GSCS_BY_Raw_Material',
i_errorcode, i_errortext, i_datetime
);

COMMIT;
end;

中级会员 注册时间: 2008-04-27 22:20:20 文章: 49 离线

兄弟，你最好把你的邏輯規則描述清楚，比如你說的中间的复杂的逻辑转换，到底是什么邏輯轉換？你給出一長串Procedure，這樣看很累的啊。。。。。。。。。

版主 注册时间: 2008-05-10 07:28:35 文章: 6 离线

用datastage的store procedure stage 实现
datastage 不是用来代替sp 的.

你这个业务逻辑有它自身的封闭性.
关于store procedure 如何使用可以参考 pdf文件

初级会员 注册时间: 2008-06-02 19:32:03 文章: 14 离线

好的，我来看看。谢谢

初级会员 注册时间: 2008-06-02 19:32:03 文章: 14 离线

在sqlserver中写这样一个procedure：
create procedure sample
as
select id,name from emp
/
在datastage中的procedure stage可以用上面的procedure做input。
但是我看了一下DB2和oracle的procedure，procedure不允许这样写，
请问DB2或oracle的procedure应该如何写才能让datastage的procedure
stage作为input。
最好做个简单例子show一下。谢谢

首先oracle和sql server存储过程的写法是不一样的，sql server那样写的存储过程可以返回结果集，但是oracle返回结果集的存储过程不能那样写，怎么写参考下面文章:
http://www.dedecms.com/web-art/shujuku/Oracle/20060913/37055.html
这是一个常见问题。

procedure stage上面有个type的下拉选项，好像是input,output,transform三项，你选input就可以作为input了。

初级会员 注册时间: 2008-06-02 19:32:03 文章: 14 离线

好的，谢谢楼主。我看看！

jerrygwang wrote:好的，谢谢楼主。我看看！

搂主是你自己！

初级会员 注册时间: 2008-06-02 19:32:03 文章: 14 离线

datastage我也是自己学习，问题是我现在导入table或者store procedure的时候提示我连接不上找不到tns，
但是我通过sqlplus就可以。
所以想问一下，是不是我再procedure中定义一个out的游标参数，然后在store procedure stage中的type
设置为source，那么这个stage就可以识别出其中的数据。
我现在正在找连接不上的问题。
所以想提前问一下lynx286 这个问题。谢谢！

如果你想要用存储过程返回一个结果集当source，建议不要这么做，datastage没这种做法，oracle也几乎不这么做。

建议从简单的mapping开始学起，而且不要以写存储过程的思维考虑datastage的问题。