[OCI]S3バケット内に追加されるCSVファイルを一定の間隔でAutonomous Databseに自動的にロードしてみた

# はじめに

Autonomous Databaseに用意されているDBMS_CLOUD_PIPELINEパッケージを使用して、S3バケットに追加されるCSVファイルをAutonomous Database内の表に自動的にロードしてみました。

# 1. AWS側の事前準備

S3バケットを作成し、以下のような内容のCSVファイルemployee1.csvをアップロードしました。

```text:employee1.csv
鈴木さん,38,1000000
田中さん,51,2500000
```

![スクリーンショット 2023-10-24 14.50.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/784824/5b6572f0-d578-c536-2032-c2d6747f1c76.png)

# 2. データのロード先となる表の作成
S3バケット内にあるCSVファイルのロード先となる表を作成します。
ここでは表の名前をemployeeとしています。
```
15:08:59 SQL> CREATE TABLE EMPLOYEE(
  2      name   VARCHAR2(128),
  3      age    NUMBER,
  4      salary NUMBER
  5* );

Table EMPLOYEEは作成されました。

15:09:19 SQL> 
```

# 3. S3にアクセスするためのクレデンシャルの作成
こちらの記事の手順を参考にして、S3バケットにアクセスするためのクレデンシャルを作成します。
ここではクレデンシャルの名前を**S3_CRED**として作成しています。

https://qiita.com/500InternalServerError/items/bf5917270c1291a3558f

```
15:09:19 SQL> BEGIN
  2    DBMS_CLOUD.CREATE_CREDENTIAL(
  3      credential_name => 'S3_CRED',
  4      params =>
  5          JSON_OBJECT('aws_role_arn' value 'arn:aws:iam::XXXXXXXXXX:role/<role_name>',          
  6                      'external_id_type' value 'database_ocid')
  7    );
  8  END;
  9* /

PL/SQLプロシージャが正常に完了しました。

15:10:12 SQL>
```
クレデンシャルの確認のために、DBMS_CLOUD.LIST_OBJECTファンクションを使用して、S3バケット内のオブジェクトの一覧を取得してみます。
```
15:10:12 SQL> SELECT object_name
  2  FROM DBMS_CLOUD.LIST_OBJECTS(
  3           credential_name => 'S3_CRED',
  4           location_uri    => 'https://s3.ap-northeast-1.amazonaws.com/バケット名/'
  5* );

OBJECT_NAME      
________________ 
employee1.csv    

15:11:19 SQL> 
```
S3バケット内のオブジェクトのリストが表示されましたので、クレデンシャルS3_CREDを使用してS3バケットにアクセスできることが確認できました。

# 4. データロードのためのパイプラインの作成
DBMS_CLOUD_PIPELINE.CREATE_PIPELINEプロシージャを使用して、S3バケット内のファイルを継続的に表にロードするパイプラインを作成します。
ここでは、S3バケット内のCSVファイルを表employeeに5分間隔で継続的にロードするパイプラインを**S3_LOAD_PIPELINE**という名前で作成しています。
```
15:12:55 SQL> BEGIN
  2      DBMS_CLOUD_PIPELINE.CREATE_PIPELINE(
  3          pipeline_name => 'S3_LOAD_PIPELINE',
  4          pipeline_type => 'LOAD',
  5          attributes    => JSON_OBJECT(
  6                               'credential_name' VALUE 'S3_CRED',
  7                               'location' VALUE 'https://s3.ap-northeast-1.amazonaws.com/s3pipelinebucket/',
  8                               'table_name' VALUE 'employee',
  9                               'format' VALUE '{"type": "csv"}',
 10                               'priority' VALUE 'MEDIUM',
 11                               'interval' VALUE '5')
 12      );
 13  END;
 14* /

PL/SQLプロシージャが正常に完了しました。

15:12:55 SQL> 
```

# 5. 作成したパイプラインの開始
DBMS_CLOUD_PIPELINE.START_PIPELINEプロシージャを使用して、パイプラインS3_LOAD_PIPELINEを開始します。
```
15:13:48 SQL> EXEC DBMS_CLOUD_PIPELINE.START_PIPELINE('S3_LOAD_PIPELINE');

PL/SQLプロシージャが正常に完了しました。

15:13:49 SQL> 
```

# 6. 動作確認

パイプラインs3_load_pipelineの動作を確認するために、表employeeのデータを確認してみます。
```
15:14:16 SQL> SELECT * FROM employee;

NAME       AGE     SALARY 
_______ ______ __________ 
鈴木さん        38    1000000 
田中さん        51    2500000 

15:14:16 SQL>
```
S3バケット内にあるemployee1.csvの内容が表employeeにロードされていることが確認できました。

S3バケットに以下のような内容のCSVファイルemployee2.csvをアップロードします。
```text:employee2.csv
渡辺さん,22,1000000
松本さん,41,3000000
```
![スクリーンショット 2023-10-24 15.15.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/784824/d0faa5c9-7266-2fe3-a7c5-1f3fdfbcebdd.png)

5分ほど待って、表employeeのデータを確認してみます。
```
15:19:01 SQL> SELECT * FROM employee;

NAME       AGE     SALARY 
_______ ______ __________ 
鈴木さん        38    1000000 
田中さん        51    2500000 
渡辺さん        22    1000000 
松本さん        41    3000000 

15:19:01 SQL>
```
S3バケット内に追加されたemployee2.csvの内容が表employeeにロードされていることが確認できました。

S3バケットに以下のような内容のCSVファイルemployee3.csvをアップロードします。
```text:employee3.csv
高橋さん,28,1800000
山口さん,33,3300000
```
![スクリーンショット 2023-10-24 15.20.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/784824/2a08ab4a-0324-2872-02fe-b83c9bbda965.png)

5分ほど待って、表employeeのデータを確認してみます。
```
15:24:58 SQL> SELECT * FROM employee;

NAME       AGE     SALARY 
_______ ______ __________ 
鈴木さん        38    1000000 
田中さん        51    2500000 
渡辺さん        22    1000000 
松本さん        41    3000000 
高橋さん        28    1800000 
山口さん        33    3300000 

6行が選択されました。 

15:24:59 SQL>
```
S3バケット内に追加されたemployee2.csvの内容が表employeeにロードされていることが確認できました。

# まとめ
DBMS_CLOUD_PIPELINEパッケージを使用して、S3バケット内に追加されるファイルをAutonomous Databseに自動的にロードできることが確認できました。

# 参考情報
・[[OCI]Autonomous Databaseのデータ・パイプライン機能（DBMS_CLOUD_PIPELINE）を試してみた（データ・ロード編）](https://qiita.com/500InternalServerError/items/daef5e28947ca75d923f)
・[[OCI]Autonomous DatabaseからAmazon S3にARNを用いたロールベースの認証でアクセスしてデータ連携してみた](https://qiita.com/500InternalServerError/items/bf5917270c1291a3558f)