Aurora Cluster Cloneを使ったDataLake連携

Aurora ClusterからのDatalake連携で、Cluster Cloneという仕組みを活用すると非常に便利だったので紹介します。

この記事では、Aurora Clusterをbatch連携でDataLakeに流し込む設計の話を扱います。他にもDMSを用いてstream連携する方法などがありますが、ここではscope outします。

## そもそも「DataLake連携」とは？

チームや書籍によって少しずつ定義が異なると思いますが、ここでは「データ分析環境を構築するために本番DBからデータを抽出してくる仕組み」を指します。

データ分析では複雑なSQLやpython scriptを用いてデータを処理していくことが多いですが、これを本番DBに直接実行してしまうと負荷が大きくサービス運用に支障が出ます。
そこで、AWSではS3 bucketにデータを吐き出すことでそこにAthenaを通してアクセスでき、負荷を考えずにデータ分析を行える仕組みがあります。

今回の話は、「S3 bucketにデータを吐き出す」までをいかにやるか？という話です。

## 基本的なDataLake連携の設計とその課題

最もシンプルな設計は本番DBに対してSQLを実行し、結果をS3 bucketにuploadする仕組みです。
SQLのclientはEC2 instanceを立てても良いですし、データ量が多くなければLambdaなどでも十分かもしれません。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109543/9f890124-cf26-0a06-4b12-c273875cb461.png)

私のチームでも、実際にこの仕組みを保守・運用していましたが、以下の欠点がありAurora Cluster Cloneを活用することに決めました。

### 本番DBに負荷がかかる

この設計では「本番DBに負荷がかかる」という欠点があります。
DBに対して何か操作するときは負荷がかかるのが当たり前ですが、DataLake連携では全てのrecordを抽出することが大事なので、一般的に事業で実行されるSQLよりも重くなりがちです。

DBの負荷を削減するために「必要最低限のTableのみを連携する」「連携のタイミングを調整する」などの対策を行い、結果データ分析の幅が狭まってしまうこともあり得ます。
DataLake連携の開発では、この「本番DBの負荷」が常に障害として立ちはだかるのです。

シンプルに技術的な問題だけであれば、Read Replicaの活用・Instance typeの変更・indexの作成などで調整できることはあります。
しかし、技術的な障壁だけでなく、開発プロセスへの影響が大きいです。
DBの負荷はサービスチームが最も気にしている箇所なので、DataLake連携の開発チームから「DBに負荷がかかるんですが、、、」と相談すると、みんな少しドキッとするでしょう。
なので早めの丁寧なコミュニケーションが求められ、結果として開発速度の低下につながります。

負荷の問題を考慮しなくて済むことで、「サービス開発チームとDataLake連携開発チームのコミュニケーションを減らす」という効果があるのです。

### 全量連携と差分連携で個別の仕組みを開発する必要がある

DataLake連携は大きく「全量連携」と「差分連携」の二つに分かれます。
「全量連携」は「今そのTableにある全てのRecordを抽出してくる」、「差分連携」は「前回抽出した時刻以降に更新されたrecordを抽出する」という機能です。
「全量連携」は「DataLake連携をデプロイした時」「障害発生時」などに活用します。

差分連携では抽出するrecord数も多くなく、サービスによっては本番DBの直接SQLを実行しても問題ないかもしれません。しかし、全量連携ではその時点でDBの存在している全てのrecordを持ってくる必要があるので、DBには大きな負荷がかかることが多いです。

そこで、全量連携と差分連携で個別の仕組みを構築することになります。
具体的には、「差分連携では上で紹介した本番DBに直接SQLを投げる仕組みを使い、全量連携をする際はSnapshotのS3 exportを利用する」などが考えられます。

すると管理対象が複数になるだけでなく、二つの異なる仕組みで連携されたデータを互いにcompatibleであることを確認しないといけないなど、開発の障壁が上がっていきます。
実際、SnapshotのS3 exportで吐き出したデータはデータ型が変わってしまうことがあり、結構神経を使います。

## Aurora Cluster Cloneを使ったDataLake連携

これらの問題を解決するために、「本番DBへの負荷なくDataLake連携する」仕組みとしてAurora Cluster Cloneの活用を行いました。

### Aurora Cluster Cloneの仕組み

Aurora Cluster CloneはDBのReplicationと同様に「あるClusterと同じ中身の別のClusterを作成する」仕組みです。
CloneではReplicationと異なり「Storageをコピーしない」ことで「Storageコストの削減」「立ち上がりの高速化」を実現しています。
二つのClusterが同一のStorageを持ちつつ独立したDBとして振る舞うために、copy-on-write protocolを使っています。

copy-on-write protocolでは、一方のDBに書き込み（`INSERT`や`UPDATE`）があったとき、そのrecordのpageのみをコピーして新しいpageを作り、そこに書き込むことで、他方のDBのデータに影響を与えないという仕組みです。


具体的にDataLake連携を例に見ていきます。
まず、本番Clusterが3つのPageを持っていたとします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109543/34091b58-00d0-d0ca-aac6-778a4202efef.png)

ここに対して、Cluster Cloneが作成されると「同じPageを参照するCluster」が作成されます。このClusterにたいしてデータ抽出用のSQLを実行します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109543/907a8b31-c8f7-24b7-536d-15d386bdf83d.png)

データ抽出中に本番Clusterにたいして`UPDATE`処理が実行されると、対処のPage（ここではPage1だと仮定します）がコピー（Page1'）され、そのページを本番Clusterが参照するようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109543/a91d86b0-a4ef-8179-6a90-70d034423ddf.png)
Clone Clusterは古いPage（Page1）を参照し続けているため、そこにはCluster Cloneを作成したタイミングの状態が残っています。

より詳細には以下のページを参照してください。
* [Cloning a volume for an Amazon Aurora DB cluster - Amazon Aurora](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/Aurora.Managing.Clone.html)

### Step Functionを使ったOrchestration

Cluster Cloneを活用する場合、データ抽出の前に「Clone Clusterの作成」「DB Instanceの作成」「DB Instanceの立ち上がりを待つ」、データ抽出の後に「DB Instanceの削除」「Clone Clusterの削除」というプロセスが必要になります。
全体のながれを管理するために、Step Functionを活用しました。

作成したStateMachineは以下のようになります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/109543/6594efcf-7042-2eaa-7e6b-7ebd46d95cd1.png)
`preprocess`はデータ抽出の細かい条件などを整理しているlambdaで、`clean-up`はエラー発生時の処理になります。
ここでは、それ以降のプロセスを紹介していきます。

#### 「Clone Clusterの作成」「DB Instanceの作成」

`create-clone-cluster`・`create-clone-instance`で「Clone Clusterの作成」「DB Instanceの作成」を行なっています。

```statemachine.json
"create-clone-cluster": {
    "Next": "create-clone-instance",
    "Parameters": {
        "DbClusterIdentifier": "[Clone Cluster名]",
        "DbSubnetGroupName": "[Clone ClusterのSubnetGroup名]",
        "RestoreType": "copy-on-write",
        "SourceDBClusterIdentifier": "[本番Cluster名]"
        "UseLatestRestorableTime": true,
        "VpcSecurityGroupIds": [
            "[CloneClusterのSecurity Group ID]"
        ]
    },
    "Resource": "arn:aws:states:::aws-sdk:rds:restoreDBClusterToPointInTime",
    "ResultPath": "$.create_clone_cluster",
    "Type": "Task"
},
```

実行するAPIは`restoreDBClusterToPointInTime`で、Cluster Clone専用のAPIではないので注意してください。
`RestoreType`に`copy-on-write`を指定することでCluster Cloneになります。

```statemachine.json
"create-clone-instance": {
  "Next": "wait-for-rds-startup",
  "Parameters": {
    "AvailabilityZone": "ap-northeast-1c",
    "DbClusterIdentifier.$": "$.create_clone_cluster.DbCluster.DbClusterIdentifier",
    "DbInstanceClass": "[DB Instance Type]",
    "DbInstanceIdentifier": "[DB Instance名]",
    "Engine": "aurora-postgresql",
  },
  "Resource": "arn:aws:states:::aws-sdk:rds:createDBInstance",
  "ResultPath": "$.create_clone_instance",
  "Type": "Task"
},
```

DB Instanceの作成では「Cluster Cloneに特有の設定」は特になく、普通の`createDBInstance` APIを呼び出しています。

#### 「DB Instanceの立ち上がりを待つ」

Clone Cluster・DB Instanceの立ち上がりには少し時間がかかります。なので、StateMachineの中で完了するまで待つ必要があります。
そこで、LambdaでDB Instanceの状態を確認し、まだ`InService`でなければ60秒待ち再度確認するというループを組みました。

```statemachine.json
"wait-for-rds-startup": {
  "Next": "retrieve-rds-status",
  "Seconds": 60,
  "Type": "Wait"
},
"retrieve-rds-status": {
  "Next": "check-rds-status",
  "Parameters": {
    "FunctionName": "[LambdaFunctionArn]",
    "Payload": {
      "DbInstanceIdentifier.$": "$.create_clone_instance.DbInstance.DbInstanceIdentifier"
    }
  },
  "Resource": "arn:aws:states:::lambda:invoke",
  "ResultPath": "$.retrieve_rds_status",
  "Type": "Task"
},
"check-rds-status": {
  "Choices": [
    {
      "Next": "run-extraction-glue-job",
      "StringEquals": "available",
      "Variable": "$.retrieve_rds_status.Payload.DBInstanceStatus"
    }
  ],
  "Default": "wait-for-rds-startup",
  "Type": "Choice"
},
```

参考：[Poll for Job Status (Lambda, AWS Batch) - AWS Step Functions](https://docs.aws.amazon.com/step-functions/latest/dg/sample-project-job-poller.html)

#### 「DB Instanceの削除」「Clone Clusterの削除」

DB Instance・Clone Clusterの削除でもClone Clusterの特殊性はありません。
通常のDB Instance, Clusterと同様に`deleteDBInstance`と`deleteDBCluster`を呼んであげれば大丈夫です。

```statemachine.json
"delete-clone-instance": {
  "Next": "delete-clone-cluster",
  "Parameters": {
    "DbInstanceIdentifier.$": "$.create_clone_instance.DbInstance.DbInstanceIdentifier",
    "SkipFinalSnapshot": true
  },
  "Resource": "arn:aws:states:::aws-sdk:rds:deleteDBInstance",
  "ResultPath": "$.delete_clone_instance",
  "Type": "Task"
},
"delete-clone-cluster": {
  "End": true,
  "Parameters": {
    "DbClusterIdentifier.$": "$.create_clone_cluster.DbCluster.DbClusterIdentifier",
    "SkipFinalSnapshot": true
  },
  "Resource": "arn:aws:states:::aws-sdk:rds:deleteDBCluster",
  "ResultPath": "$.delete_clone_cluster",
  "Type": "Task"
},
```

## 補足

### Storageを共有しているが、本当に本番DBに影響はないの？

厳密には、Storageレベルでの影響はあると思います。
Clone元のInstanceとClone Instanceは同一のStorage領域にアクセスするので、完全に負荷を0にすることはできていないでしょう。

しかし、現実的には問題にはならないと思います。DataLake連携実行時の本番DBのmetricsを確認しましたが、確認した全てのmetricsで反応はありませんでした。
AWSのStorageの分散化やアクセス方法の工夫で影響をほぼ0にできているのではないかと思っています。