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などでも十分かもしれません。
私のチームでも、実際にこの仕組みを保守・運用していましたが、以下の欠点があり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を持っていたとします。
ここに対して、Cluster Cloneが作成されると「同じPageを参照するCluster」が作成されます。このClusterにたいしてデータ抽出用のSQLを実行します。
データ抽出中に本番ClusterにたいしてUPDATE処理が実行されると、対処のPage（ここではPage1だと仮定します）がコピー（Page1'）され、そのページを本番Clusterが参照するようになります。
Clone Clusterは古いPage（Page1）を参照し続けているため、そこにはCluster Cloneを作成したタイミングの状態が残っています。
より詳細には以下のページを参照してください。
Step Functionを使ったOrchestration
Cluster Cloneを活用する場合、データ抽出の前に「Clone Clusterの作成」「DB Instanceの作成」「DB Instanceの立ち上がりを待つ」、データ抽出の後に「DB Instanceの削除」「Clone Clusterの削除」というプロセスが必要になります。
全体のながれを管理するために、Step Functionを活用しました。
作成したStateMachineは以下のようになります。
preprocessはデータ抽出の細かい条件などを整理しているlambdaで、clean-upはエラー発生時の処理になります。
ここでは、それ以降のプロセスを紹介していきます。
「Clone Clusterの作成」「DB Instanceの作成」
create-clone-cluster・create-clone-instanceで「Clone Clusterの作成」「DB Instanceの作成」を行なっています。
"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になります。
"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秒待ち再度確認するというループを組みました。
"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
「DB Instanceの削除」「Clone Clusterの削除」
DB Instance・Clone Clusterの削除でもClone Clusterの特殊性はありません。
通常のDB Instance, Clusterと同様にdeleteDBInstanceとdeleteDBClusterを呼んであげれば大丈夫です。
"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にできているのではないかと思っています。