はじめに
OCI Data Integration 増分ロードの実行（Object Storage編）に続いて、データベースをソースにして増分ロードの実行を試してみます。データベースの増分ロードを行うにはリレーショナル・データベースがソースで、増分ロードするデータを判別するためのウォーターマーク列（DATE型、DATETIME型、TAMESTAMP型のいずれか）があることが前提です。Object Storage編ではデータ・フローの統合タスクで実行しましたが、今回はデータ・ローダー・タスクで実行します。
増分ロードのソーステーブルの準備
テスト表はemp表の構成にウォーターマーク列としてWMARK_COL列を追加して作成しました。次にemp表からdeptno=10のデータを投入します。WMARK_COL列はNULLです。
DESC INC_LOAD_EMP
名前 Nullかどうか タイプ
--------- -------- ------------
EMPNO NUMBER
ENAME VARCHAR2(64)
JOB VARCHAR2(64)
MGR NUMBER
HIREDATE DATE
SAL NUMBER
COMM NUMBER
DEPTNO NUMBER
WMARK_COL DATE
insert into inc_load_emp select empno,ename,job,mgr,hiredate,sal,comm,deptno,null from emp where deptno=10;
EMPNO ENAME JOB MGR HIREDATE SAL COMM DEPTNO WMARK_COL
----- ------ --------- ---- ------------------ ---- ---- ------ ---------
7934 MILLER CLERK 7782 1982/1/23 0:00:00 1300 10
7782 CLARK MANAGER 7839 1981/6/9 0:00:00 2450 10
7839 KING PRESIDENT 1981/11/17 0:00:00 5000 10
OCI Data Integrationでデータ・ローダー・タスクを作成
データ・ローダー・タスクの作成ウィザードに沿って設定していきます。
ソースとターゲットのタイプを指定します。ソースはデータベース、ターゲットはObject Storageなのでファイル・ストレージを選択します。
ソースの設定をします。あらかじめ作成済みのAutonomous Databaseのデータ・アセットから、スキーマとテスト表を指定します。
設定タブで増分ロードにチェックを、ウォーターマーク列にWMARK_COL列を指定します。
ターゲットの設定をします。今回は1つのファイルとして出力するように出力を単一ファイルとして作成にチェックをし、エンティティ名の項目に出力ファイル名を指定しました。
出力を単一ファイルとして作成にチェックしない場合は出力先のディレクトリを指定します。統合戦略が挿入の場合はディレクトリに新しく出力ファイルが追加され、上書きの場合は、ディレクトリを上書きするので既存の出力ファイルは削除され新しい出力ファイルのみとなります。
設定内容を確認して、作成して閉じるをクリックして、タスクの作成は完了です。
アプリケーションに公開して実行する
既存のアプリケーションに公開して実行します。Object Storage編と同様にDATETIMEとしてSYS.LAST_LOAD_DATEというパラメータが使用されています。そのまま実行してみます。
結果を確認すると、成功はしていますが、書き込まれた行に値がなくロードはされてなさそうです。
ターゲットに指定したバケットにファイルは生成されていました。しかし、内容を確認するとヘッダーだけでデータは出力はされていませんでした。
WMARK_COLにデータとタイムスタンプを入れて実行
次にテスト表にDEPTNO=20のデータをWMARK_COL列に現在の時刻を入れます。
insert into inc_load_emp select empno,ename,job,mgr,hiredate,sal,comm,deptno,current_date from emp where deptno=20;
select * from inc_load_emp;
EMPNO ENAME JOB MGR HIREDATE SAL COMM DEPTNO WMARK_COL
----- ------ --------- ---- ------------------ ---- ---- ------ ------------------
7369 SMITH CLERK 7902 1980/12/17 0:00:00 800 20 2023/12/18 5:19:39
7876 ADAMS CLERK 7788 1983/1/12 0:00:00 1100 20 2023/12/18 5:19:39
7566 JONES MANAGER 7839 1981/4/2 0:00:00 2975 20 2023/12/18 5:19:39
7788 SCOTT ANALYST 7566 1982/12/9 0:00:00 3000 20 2023/12/18 5:19:39
7902 FORD ANALYST 7566 1981/12/3 0:00:00 3000 20 2023/12/18 5:19:39
7934 MILLER CLERK 7782 1982/1/23 0:00:00 1300 10
7782 CLARK MANAGER 7839 1981/6/9 0:00:00 2450 10
7839 KING PRESIDENT 1981/11/17 0:00:00 5000 10
データ・ローダー・タスクを実行すると、SYS.LAST_LOAD_DATEには前に実行したタスクの時間がはいっています。
タスクの実行は成功、結果は5行追加されています。
バケットのファイルを確認すると、追加したDEPTNO=20のデータが5行出力されています。
なお、アプリケーションの実行画面からタスクの再実行を選択することができますが、その場合SYS.LAST_LOAD_DATEは前回のタスク実行時の値（今回の場合1970-01-01 00:00:00）になります。
タスクを修正して再公開して実行
数値型のデータが整数値として出力されていないため、データ型の変換を行う修正をタスクに行います。プロジェクトのタスクから作成したデータ・ローダー・タスクをクリックします。
上部にあるメニューから④変換オプションを選択します。
データのフィルタでNUMERICを選択するとNUMERICの列だけ表示がされます。
アクションからデータ型の変更を選択し、INTEGER型に変換して適用します。
変換されていることを確認して（表示が変わらない場合は別のターゲットなどの画面に切り替えて戻る）、データ・ローダー・タスクを保存します。
修正したタスクを再度アプリケーションに公開して実行します。SYS.LAST_LOAD_TIMEが最初の値に戻りました。再公開とともにリセットされるようです。
このまま実行してもDEPTNO=20のデータは入るはずなので実行してみます。予想通り実行結果は5行ロードされました。
単一ファイルの場合は、上書きになるので同じ名前のファイルを確認します。
NUMBER型のデータも整数で表示されています。もしDATE型のフォーマットを指定したい場合もタスク作成の変換で設定することで可能です。
SYS.LAST_LOAD_DATEを手動変更して実行してみる
DEPTNO=30のデータを新規に表に挿入しました。
insert into inc_load_emp select empno,ename,job,mgr,hiredate,sal,comm,deptno,current_date from emp where deptno=30;
select * from inc_load_emp;
EMPNO ENAME JOB MGR HIREDATE SAL COMM DEPTNO WMARK_COL
----- ------ --------- ---- ------------------ ---- ---- ------ ------------------
7369 SMITH CLERK 7902 1980/12/17 0:00:00 800 20 2023/12/18 5:19:39
7876 ADAMS CLERK 7788 1983/1/12 0:00:00 1100 20 2023/12/18 5:19:39
7566 JONES MANAGER 7839 1981/4/2 0:00:00 2975 20 2023/12/18 5:19:39
7788 SCOTT ANALYST 7566 1982/12/9 0:00:00 3000 20 2023/12/18 5:19:39
7902 FORD ANALYST 7566 1981/12/3 0:00:00 3000 20 2023/12/18 5:19:39
7900 JAMES CLERK 7698 1981/12/3 0:00:00 950 30 2023/12/18 8:24:27
7521 WARD SALESMAN 7698 1981/2/22 0:00:00 1250 500 30 2023/12/18 8:24:27
7654 MARTIN SALESMAN 7698 1981/9/28 0:00:00 1250 1400 30 2023/12/18 8:24:27
7844 TURNER SALESMAN 7698 1981/9/8 0:00:00 1500 0 30 2023/12/18 8:24:27
7499 ALLEN SALESMAN 7698 1981/2/20 0:00:00 1600 300 30 2023/12/18 8:24:27
7698 BLAKE MANAGER 7839 1981/5/1 0:00:00 2850 30 2023/12/18 8:24:27
7934 MILLER CLERK 7782 1982/1/23 0:00:00 1300 10
7782 CLARK MANAGER 7839 1981/6/9 0:00:00 2450 10
7839 KING PRESIDENT 1981/11/17 0:00:00 5000 10
SYS.LAST_LOAD_DATEを手動で修正してみます。2023/12/18 8:00:00に設定して実行しました。
成功し6行書き込まれました。
内容を確認するとDEPTNO=30のデータが出力されていました。SYS.LAST_LOAD_DATE以降のデータが出力されていることがわかります。
おわりに
OCI Data Integrationでデータベースをソースにした増分ロードを試してみました。データが変更されたときのタイムスタンプなどのデータがあれば、ウォーターマーク列として使用することで、前のタスク実行時間以降のデータのみをロードさせることができます。今回はデータ・ローダー・タスクで実行しましたが、データ・フローでも同様に可能です。