OCI Data Integration増分ロードの実行(Oracle Database編)

# はじめに
[OCI Data Integration 増分ロードの実行（Object Storage編）](https://qiita.com/goo2000/items/4751f91d390d71560228)に続いて、データベースをソースにして増分ロードの実行を試してみます。データベースの増分ロードを行うにはリレーショナル・データベースがソースで、増分ロードするデータを判別するためのウォーターマーク列（DATE型、DATETIME型、TAMESTAMP型のいずれか）があることが前提です。Object Storage編ではデータ・フローの統合タスクで実行しましたが、今回はデータ・ローダー・タスクで実行します。

# 増分ロードのソーステーブルの準備
テスト表はemp表の構成にウォーターマーク列としてWMARK_COL列を追加して作成しました。次にemp表からdeptno=10のデータを投入します。WMARK_COL列はNULLです。

```sql
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なのでファイル・ストレージを選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/f15c4be8-bd11-428a-df85-b932389ef0d0.png)

ソースの設定をします。あらかじめ作成済みのAutonomous Databaseのデータ・アセットから、スキーマとテスト表を指定します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/253cf7a8-4fa2-cadb-24b7-d8235642046a.png)

設定タブで**増分ロード**にチェックを、**ウォーターマーク列**にWMARK_COL列を指定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/bc5fa727-4e7a-86aa-071e-16b192372d04.png)

ターゲットの設定をします。今回は1つのファイルとして出力するように**出力を単一ファイルとして作成**にチェックをし、エンティティ名の項目に出力ファイル名を指定しました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/d99e53e4-803d-3c0e-4cf4-4fd19c0004cc.png)

::: note
出力を単一ファイルとして作成にチェックしない場合は出力先のディレクトリを指定します。統合戦略が挿入の場合はディレクトリに新しく出力ファイルが追加され、上書きの場合は、ディレクトリを上書きするので既存の出力ファイルは削除され新しい出力ファイルのみとなります。
:::

設定内容を確認して、**作成して閉じる**をクリックして、タスクの作成は完了です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/e727ada0-06d9-38e9-daa4-68eadc84b8f2.png)

# アプリケーションに公開して実行する
既存のアプリケーションに公開して実行します。Object Storage編と同様にDATETIMEとして**SYS.LAST_LOAD_DATE**というパラメータが使用されています。そのまま実行してみます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/2aec80d8-af81-3789-e8bd-45a74e45fb4e.png)

結果を確認すると、成功はしていますが、書き込まれた行に値がなくロードはされてなさそうです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/10c72172-c864-3b05-7fa3-ec73903e65cf.png)

ターゲットに指定したバケットにファイルは生成されていました。しかし、内容を確認するとヘッダーだけでデータは出力はされていませんでした。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/20aa0f0e-1c31-9fa5-d5e9-54e3304df171.png)

# WMARK_COLにデータとタイムスタンプを入れて実行
次にテスト表にDEPTNO=20のデータをWMARK_COL列に現在の時刻を入れます。
```sql
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には前に実行したタスクの時間がはいっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/bbc2bc84-a9d8-a49f-8fc9-a7418e0afcc0.png)

タスクの実行は成功、結果は5行追加されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/dd79e81d-7b42-31fa-0ab6-60c743c00a21.png)

バケットのファイルを確認すると、追加したDEPTNO=20のデータが5行出力されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/1606eba4-d4bb-2a99-7770-8f78d3259768.png)

::: note
なお、アプリケーションの実行画面からタスクの再実行を選択することができますが、その場合SYS.LAST_LOAD_DATEは前回のタスク実行時の値（今回の場合1970-01-01 00:00:00）になります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/1715fe80-f6eb-e02c-f9f0-e9d9156724eb.png)
::: 

# タスクを修正して再公開して実行
数値型のデータが整数値として出力されていないため、データ型の変換を行う修正をタスクに行います。プロジェクトのタスクから作成したデータ・ローダー・タスクをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/ba5c2d26-0c43-dc98-cec9-fa45f35ce039.png)

上部にあるメニューから**④変換オプション**を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/8ed4f4b0-6c0f-9413-25ca-bb8a280f7fd6.png)

**データのフィルタ**でNUMERICを選択するとNUMERICの列だけ表示がされます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/96c3e52f-66a8-dbfe-7a0f-55df96fc1f79.png)

**アクション**から**データ型の変更**を選択し、INTEGER型に変換して適用します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/5aa07468-40e9-b1ce-a7f2-178cfff51678.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/38dd9870-8a61-d205-cdf4-4c66745e05c2.png)

変換されていることを確認して（表示が変わらない場合は別のターゲットなどの画面に切り替えて戻る）、データ・ローダー・タスクを保存します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/841684cf-b46f-2b59-a22a-a01eeba27c6b.png)


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/30f4ec11-c184-5645-45e4-e08f6508b573.png)

修正したタスクを再度アプリケーションに公開して実行します。SYS.LAST_LOAD_TIMEが最初の値に戻りました。再公開とともにリセットされるようです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/8f271cd5-f148-cb2f-379c-0e260dca4421.png)

このまま実行してもDEPTNO=20のデータは入るはずなので実行してみます。予想通り実行結果は5行ロードされました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/1ad09c2c-fa95-323f-18c3-95609952e5ea.png)

単一ファイルの場合は、上書きになるので同じ名前のファイルを確認します。
NUMBER型のデータも整数で表示されています。もしDATE型のフォーマットを指定したい場合もタスク作成の変換で設定することで可能です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/517cddb6-1510-d635-718b-96ad996de8ae.png)


# SYS.LAST_LOAD_DATEを手動変更して実行してみる
DEPTNO=30のデータを新規に表に挿入しました。


```sql
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に設定して実行しました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/44e0622d-56ea-6625-a8aa-741c0268efab.png)

成功し6行書き込まれました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/807450e7-c54e-500f-1e35-f35db50ef0a6.png)

内容を確認するとDEPTNO=30のデータが出力されていました。SYS.LAST_LOAD_DATE以降のデータが出力されていることがわかります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/134317/5460d369-8c06-414b-36a8-9542571b07fd.png)


# おわりに
OCI Data Integrationでデータベースをソースにした増分ロードを試してみました。データが変更されたときのタイムスタンプなどのデータがあれば、ウォーターマーク列として使用することで、前のタスク実行時間以降のデータのみをロードさせることができます。今回はデータ・ローダー・タスクで実行しましたが、データ・フローでも同様に可能です。

# 参考資料
+ [データ・ローダー・タスクの作成](https://docs.oracle.com/ja-jp/iaas/data-integration/using/creating-a-data-loader-task.htm)