Db2→AWS Database Migration Service(DMS)→S3→Snowflakeのデータ連携

# はじめに

AWS Database Migration Service(以下、DMS)はデータソースに対する更新をキャプチャーし、更新差分データをターゲットに伝搬することができるAWSサービスです。ここではDMSのデータソースをDb2、ターゲットをS3にしてSnowflake上のテーブルへのロードを試します。

DMSからS3への連携はこちらの記事（[AWS Database Migration Service(DMS)でDb2をS3に一括移行＆(CDC的な)継続的レプリケーションする](https://qiita.com/ysskjck/items/8846db065a6c03f65565)）で確認済みです。
ここではDMSから配置されたS3上のオブジェクトをSnowflakeに反映する部分にフォーカスしています。


# 前提

実施環境として、データソースとなるDb2はCloud Pak for Data上にデプロイされたものを使用しました。

    Cloud Pak for Data：Version 4.7
    Openshift：Version 4.12
    Cloud：IBM Cloud
    （外部公開用のPublic Load Balancer IPを使用して接続）

[こちら](https://docs.aws.amazon.com/ja_jp/dms/latest/userguide/CHAP_Introduction.Sources.html#CHAP_Introduction.Sources.title)を見ると2024年3月1日現在DMSがデータソースとして対応しているDb2のバージョンが確認できます。

    - IBM DB 2 for Linux、UNIX、および Windows (Db2 LUW) 
        - バージョン 9.7、すべてのフィックスパック
        - バージョン 10.1、すべてのフィックスパック
        - バージョン 10.5、フィックスパック 5 を除くすべてのフィックスパック
        - バージョン 11.1、すべてのフィックスパック
        - バージョン 11.5 Mods (1～8)、フィックスパック 0 のみ
    - IBM Db2 for z/OS
        - バージョン 12
    - Amazon RDS for IBM Db2 LUW

用意したDb2環境は上記の対応製品・バージョンに該当していないような気もしましたが、そこはあまりこだわらず試しています。

参考：[DMSのAWS公式](https://aws.amazon.com/jp/dms/)

# 実施手順

- SnowflakeからS3を外部ステージとして使用できるようにする
Showflake公式の [Amazon S3へのセキュアアクセスの構成](https://docs.snowflake.com/ja/user-guide/data-load-s3-config) を参考に実施します。
ここでは以下の3つの方法について記載があります。

      オプション1：Amazon S3にアクセスするためのSnowflakeストレージ統合の構成
      オプション2：Amazon S3にアクセスするための AWS IAM ロールの設定 --- 廃止
      オプション3：Amazon S3にアクセスするための AWS IAM ユーザー認証情報の設定

    オプション1が推奨とされています。
    オプション2は廃止となっています。
    今回は、セキュリティ面ではオプション1が良いと考えられますが、楽をとってオプション3で実施します。
    （既存のS3アクセス権限のあるIAMユーザーの認証情報を使用する）

- S3のファイルをSnowflakeのテーブルに反映する

# やってみた

- SnowflakeからS3を外部ステージとして使用できるようにする

```sql:外部ステージ作成
-- データベース、スキーマの指定
USE SCHEMA MY_DATABASE.PUBLIC;

-- 外部ステージを定義（*****は認証情報をマスキング）
CREATE OR REPLACE STAGE my_S3_stage
  URL='s3://yssk-mybacket/snowflake-test/'
  CREDENTIALS=(AWS_KEY_ID='*****' AWS_SECRET_KEY='*****')
  ENCRYPTION=(TYPE='AWS_SSE_KMS' KMS_KEY_ID = 'aws/key');
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/f1138286-8dcc-c81e-ed13-3e5c1b4f4abf.png)
（指定したCREDENTIALSはS3へのアクセス権限があるIAMユーザーのもの）

外部ステージとして使用するS3フォルダの内容
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/d98dc4ba-7032-49f1-d856-c1d2a41038c9.png)

Snowsightで外部ステージを確認
```SQL
list @my_S3_stage;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/c3f31b71-e6d3-fb86-bdf0-7296499d6055.png)

```SQL:DMSから連携される断面データ（初期移行データの想定）の内容確認
SELECT t.$1, t.$2, t.$3 FROM @my_S3_stage t WHERE t.$1 NOT IN ('I','U','D');
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/9cb432f1-3740-aed4-1a2f-6351223fcd1a.png)

```SQL:DMSから連携される更新差分データの内容確認
SELECT t.$1, t.$2, t.$3, t.$4 FROM @my_S3_stage t WHERE t.$1 IN ('I','U','D');
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/b168ed7f-1571-8e83-12ce-32191bc7b574.png)

- S3のファイルをSnowflakeのテーブルに反映する
```SQL:DMSから連携される断面データ（初期移行データの想定）のロード
COPY INTO MY_DATABASE.PUBLIC.MY_LOAD_TEST FROM
(SELECT t.$1, t.$2, t.$3 FROM @my_S3_stage/LOAD00000001.csv t);
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/72fc4e38-9429-103f-bc7b-fb3afb6c5bef.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/115f46fd-9eda-552d-7fe9-644fd8e6b1cb.png)

```SQL:DMSから連携されるINSERT分の反映
INSERT INTO MY_DATABASE.PUBLIC.MY_LOAD_TEST (NO, NAME, TEXT) 
SELECT t.$2, t.$3, t.$4 FROM @my_S3_stage t WHERE t.$1 = 'I';
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/e0813e28-402b-8662-3a7c-2e47216e3313.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/c7bdd717-e56b-6117-2fcd-6cc01e1e0de0.png)

```SQL:DMSから連携されるUPDATE分の反映
MERGE INTO MY_DATABASE.PUBLIC.MY_LOAD_TEST a
USING (
  SELECT
    $2 AS NO,
    $3 AS NAME,
    $4 AS TEXT
  FROM
    @my_S3_stage t
  WHERE
    $1 = 'U'
) t
ON (
  a.NO = t.NO
)
WHEN MATCHED THEN
  UPDATE SET
    a.NAME = t.NAME,
    a.TEXT = t.TEXT;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/44b4a3e8-834c-dbe9-36fc-48ebf9ab521e.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/9c3c040a-3a59-5119-c16b-2284b7670905.png)


```SQL:DMSから連携されるDELETE分の反映
MERGE INTO MY_DATABASE.PUBLIC.MY_LOAD_TEST a
USING (
  SELECT
    $2 AS NO,
    $3 AS NAME,
    $4 AS TEXT
  FROM
    @my_S3_stage t
  WHERE
    $1 = 'D'
) t
ON (
  a.NO = t.NO
)
WHEN MATCHED THEN
  DELETE;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/aa5beff2-8563-a6a0-5fde-f1f721a61d86.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/838636/2048e955-04ba-3ce4-c7e9-3e9eb1fc1474.png)


# まとめ

やりたいことは確認できました。

実運用ではタスクを設定して一定間隔でINSERT・UPDATE・DELETE分を随時反映させたり、Snowpipeによるストリーミング処理を実装してS3へのオブジェクト配置をトリガーに即ロードさせたりする必要があるのではないかと思います。
また、エラーハンドリングなども必要になります。

（本稿を作成中に発見した [クラスメソッド様の記事：AWS DMSから出力されたファイルをSnowflakeにロードしてみた](https://dev.classmethod.jp/articles/try-loading-the-file-output-from-aws-dms-into-snowflake/) がSnowpipeを使用していて非常に参考になります）

取っ掛かりとして、参考になればと思います。