Aurora MySQL の変更データを AWS DMS の Full Load + CDC で S3（Parquet）に出力し Athena でクエリしてみた

はじめに

テクニカルサポートの 片方 です。

Aurora MySQL のデータを分析用途で Amazon S3 に連携する方法の 1 つとして、AWS Database Migration Service（AWS DMS）を利用した Full Load + Change Data Capture（CDC）があります。

本ブログでは、AWS 環境がない状態から Aurora MySQL Serverless v2、AWS DMS、Amazon S3、Amazon Athena を構築し、Aurora MySQL の既存データと変更データを S3 に Parquet 形式で出力します。構成は以下のとおりです。

Aurora MySQL
  └─ AWS DMS
       └─ Amazon S3（Parquet）
            └─ Amazon Athena

AWS DMS では、既存データを Full Load として取り込み、その後の変更を CDC として継続的に取得できます。S3 ターゲットでは Parquet 形式を選択でき、CDC の操作種別として I（INSERT）、U（UPDATE）、D（DELETE）を出力できます。

https://docs.aws.amazon.com/ja_jp/dms/latest/userguide/CHAP_Target.S3.html

本ブログで確認する内容は以下です。

• Aurora MySQL をソースとした AWS DMS の Full Load + CDC の構築
• Amazon S3 への Parquet 形式でのデータ出力
• INSERT、UPDATE、DELETE の変更データ出力
• Amazon Athena から raw CDC データをクエリする方法

なお、本ブログで Athena から参照するデータは、変更履歴を保持する raw データです。同一レコードへの複数回の更新や削除イベントを含むため、そのまま最新状態のテーブルとして利用することは想定していません。最新状態のテーブルを作成するには、後続処理で操作種別や主キー、変更順序を考慮したデータ変換が必要です。

https://docs.aws.amazon.com/ja_jp/dms/latest/sbs/mysql-s3datalake.stepbystep.html

また、Aurora MySQL で CDC を利用するには、バイナリログを有効化し、binlog_format を ROW、binlog_row_image を full に設定する必要があります。本ブログでは検証用の Aurora MySQL Serverless v2 を使用します。Aurora MySQL Serverless v2 は AWS DMS の CDC をサポートしています。

https://docs.aws.amazon.com/ja_jp/dms/latest/userguide/CHAP_Source.MySQL.html

構築は AWS マネジメントコンソールで行い、検証完了後は不要な課金を避けるため、作成したリソースを削除します。

検証環境

Aurora MySQL の既存データと変更データを AWS DMS で Amazon S3 に出力し、Amazon Athena で確認する環境を構築します。構成は以下のとおりです。

AWS DMS は、Aurora MySQL の既存データを Full Load で出力した後、継続的な変更を CDC として Amazon S3 に出力します。出力形式には Parquet を使用し、AWS Glue Data Catalog に登録したテーブルを Amazon Athena からクエリします。

使用する AWS サービス

検証環境の概要

Aurora MySQL と AWS DMS Replication Instance は同一 VPC に配置します。DMS から Aurora MySQL へ接続できるようにセキュリティグループを設定し、DMS から Amazon S3 へデータを出力します。

検証用テーブル

CDC の動作を確認するため、Aurora MySQL に注文情報を管理する app.orders テーブルを作成します。

本ブログでは、初期データを Full Load で出力した後に app.orders へ INSERT、UPDATE、DELETE を実行し、S3 に出力された CDC データを Amazon Athena で確認します。

事前準備

今回は以下の名前で進めます。

VPC、Private Subnet を作成する

VPC を作成します。

DMS Replication Subnet Group には、同一 VPC 内かつ異なる AZ にある少なくとも 2 つのサブネットが必要です。

https://docs.aws.amazon.com/ja_jp/dms/latest/userguide/CHAP_ReplicationInstance.VPC.html

S3 用の Gateway VPC Endpoint を作成する

NAT Gateway を作成せずに Private Subnet 上の DMS から S3 へアクセスするため、S3 用の Gateway 型 VPC エンドポイントを作成します。

セキュリティグループを作成する

以下のインバウンドルールを追加します。

DMS Replication Instance と Aurora MySQL の両方に同じセキュリティグループを割り当てます。セキュリティグループ自身をソースに指定することで、同じセキュリティグループを持つリソース間の MySQL 接続だけを許可できます。

実装してみた

ここからは、CDC のソースとなる Aurora MySQL Serverless v2 を作成し、AWS DMS で変更データを取得するための設定を行います。

Aurora MySQL で CDC を有効化する

AWS DMS で Aurora MySQL の変更データを取得するには、バイナリログを使用できるように設定します。Aurora MySQL Serverless v2 は AWS DMS の CDC をサポートしています。一方、Serverless v1 はバイナリログを利用できないため CDC をサポートしていません。

DB クラスターパラメータグループを作成する

binlog_format と binlog_row_image は DB クラスターパラメータグループでのみ設定できるため、カスタム DB クラスターパラメータグループを作成します。

作成後、以下のパラメータを変更しました。

ROW を設定することで行単位の変更がバイナリログに記録されます。full を設定すると変更前後のすべての列値が記録され、AWS DMS が UPDATE と DELETE を正しく再現できます。

Aurora MySQL Serverless v2 を作成する

CDC のソースとなる Aurora MySQL Serverless v2 クラスターを作成しました。主な設定は以下のとおりです。

本ブログは検証用途のため、Aurora レプリカや RDS Proxy、Backtrack、拡張モニタリング、CloudWatch Logs へのログエクスポート、削除保護は有効化していません。ストレージ暗号化には AWS 所有 KMS キーを使用しました。

Aurora MySQL は Private Subnet に配置しました。後続で作成する AWS DMS Replication Instance には、同じ VPC 内から MySQL のポート 3306 で接続させます。

また、RDS Data API を有効化しました。これにより、踏み台サーバーや VPN を用意せず、クエリエディタから SQL を実行できます。

https://docs.aws.amazon.com/ja_jp/AmazonRDS/latest/AuroraUserGuide/query-editor.html

DMS 接続用ユーザーを作成する

クラスターと Writer インスタンスのステータスがどちらも Available になったことを確認したら、クエリエディタから Aurora MySQL に接続し、AWS DMS のソースエンドポイントで使用するデータベースユーザーを作成します。

以下の SQL を実行しました。<DMS_USER_PASSWORD> には、検証用に作成したパスワードを指定します。

CREATE USER 'dms_user'@'%'
IDENTIFIED BY '<DMS_USER_PASSWORD>';

GRANT SELECT ON app.* TO 'dms_user'@'%';

GRANT REPLICATION CLIENT, REPLICATION SLAVE ON *.*
TO 'dms_user'@'%';

付与した権限の役割は以下のとおりです。

MySQL 8.0 系では動的権限が追加されていますが、REPLICATION SLAVE は引き続き有効です。ここでは AWS DMS 公式ドキュメントに記載された必要権限に従っています。

権限を付与した後、以下の SQL で設定内容を確認します。

SHOW GRANTS FOR 'dms_user'@'%';

app データベースへの SELECT 権限と、REPLICATION CLIENT、REPLICATION SLAVE 権限が付与されていることを確認できました。

binlog の保持時間を設定する

AWS DMS が停止している間に必要なバイナリログが削除されると、CDC を継続できなくなる場合があります。今回は検証用として、binlog の保持時間を 24 時間に設定しました。

CALL mysql.rds_set_configuration(
  'binlog retention hours',
  24
);

設定内容は以下の SQL で確認できます。

CALL mysql.rds_show_configuration;

本番運用では、DMS タスクが停止した場合でも再開できるよう「想定される最大停止時間 + 余裕時間」を基準に保持時間を設定します。保持期間を超えると CDC を再開できず、Full Load からのやり直しが必要になります。

https://docs.aws.amazon.com/ja_jp/dms/latest/userguide/CHAP_Source.MySQL.html

検証用テーブルと初期データを作成する

app データベースに orders テーブルを作成します。主キーとして order_id を定義しておくことで、どのレコードに変更が発生したかを追跡しやすくなります。

CREATE TABLE app.orders (
  order_id BIGINT NOT NULL,
  customer_name VARCHAR(100) NOT NULL,
  status VARCHAR(30) NOT NULL,
  amount DECIMAL(12, 2) NOT NULL,
  updated_at DATETIME NOT NULL,
  PRIMARY KEY (order_id)
);

続けて、Full Load の対象となる初期データを 2 件登録します。

INSERT INTO app.orders (
  order_id,
  customer_name,
  status,
  amount,
  updated_at
)
VALUES
  (1, 'Alice', 'CREATED', 1200.00, NOW()),
  (2, 'Bob', 'CREATED', 2500.00, NOW());

登録したデータを確認します。

SELECT
  order_id,
  customer_name,
  status,
  amount,
  updated_at
FROM app.orders
ORDER BY order_id;

order_id が 1 と 2 の 2 件が返れば、初期データの準備は完了です。

Amazon S3 バケットを作成する

AWS DMS のターゲットとして使用する S3 バケットを作成します。AWS DMS が出力する Parquet ファイルと、Amazon Athena のクエリ結果を同じバケット内の別プレフィックスに保存します。

s3://<bucket-name>/
├── raw/
│   └── dms/
└── athena-results/

raw/dms/ は AWS DMS の出力先で、後続で作成する S3 ターゲットエンドポイントのバケットフォルダとして指定します。athena-results/ は Amazon Athena のクエリ結果出力先です。両者を分けることで、後続の AWS Glue Crawler が不要なファイルをテーブル定義に取り込むことを防ぎます。

主な設定は以下のとおりです。

バケットの作成後、以下のプレフィックスを作成しました。

raw/dms/
athena-results/

DMS 用の IAM ロールを作成する

AWS DMS が Amazon S3 に Parquet ファイルを書き込むための IAM ロールを作成します。信頼ポリシーでは、AWS DMS のサービスプリンシパルである dms.amazonaws.com を許可します。

• ロール名: dms-cdc-lab-s3-role

信頼ポリシー

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "dms.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

S3 バケット全体への過剰な権限は付与せず、出力先である raw/dms/ プレフィックスに限定したインラインポリシーを追加します。<BUCKET_NAME> は作成した S3 バケット名に置き換えてください。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListTargetBucket",
      "Effect": "Allow",
      "Action": [
        "s3:GetBucketLocation",
        "s3:ListBucket"
      ],
      "Resource": "arn:aws:s3:::<BUCKET_NAME>"
    },
    {
      "Sid": "WriteDmsRawData",
      "Effect": "Allow",
      "Action": [
        "s3:AbortMultipartUpload",
        "s3:DeleteObject",
        "s3:GetObject",
        "s3:ListMultipartUploadParts",
        "s3:PutObject",
        "s3:PutObjectTagging"
      ],
      "Resource": "arn:aws:s3:::<BUCKET_NAME>/raw/dms/*"
    }
  ]
}

AWS DMS のソース・ターゲットエンドポイントを作成する

AWS DMS では、データの取得元と出力先をエンドポイントとして登録します。

• Source endpoint: Aurora MySQL
• Target endpoint: Amazon S3

Replication Instance を作成する前でもエンドポイントは作成できますが、接続テストには Replication Instance が必要です。

Aurora MySQL のソースエンドポイントを作成する

DMS コンソールで［移行または複製］から［エンドポイント］を開き、［エンドポイントの作成］を選択しました。主な設定は以下のとおりです。

サーバー名には、インスタンスエンドポイントではなくクラスターエンドポイントを指定しました。

<cluster-identifier>.cluster-xxxxxxxxxxxx.ap-northeast-1.rds.amazonaws.com

クラスターエンドポイントは書き込み可能なプライマリ DB インスタンスへ接続するエンドポイントです。今回は Writer 1 台の構成ですが、クラスターエンドポイントを指定することで、フェイルオーバーが発生しても昇格した新しい Writer に接続されます。

なお、MySQL 互換エンドポイントではエンドポイント作成時にデータベース名を明示せず、対象データベースとテーブルは後続の DMS タスクのテーブルマッピングで指定します。

Amazon S3 のターゲットエンドポイントを作成する

続いて、S3 ターゲットエンドポイントを作成します。主な設定は以下のとおりです。

デフォルトでは S3 ターゲットは CSV 形式で出力されます。今回は Amazon Athena からクエリすることを想定し、列指向形式である Parquet を指定します。あわせて、CDC データを JST の時間単位でパーティション化するため、以下のエンドポイント設定を追加しました。

CdcInsertsOnly は既定値が false のため明示は必須ではありませんが、意図を明確にするために設定しています。

AWS DMS Replication Instance を作成する

Replication Instance は、Aurora MySQL からデータを読み取り、Amazon S3 へ出力する DMS の実行基盤です。移行処理に加えて、CDC で必要となるトランザクションログの読み取りや一時的なキャッシュも行います。

Replication Subnet Group を作成する

DMS コンソールの［サブネットグループ］から［サブネットグループを作成］を選択しました。

Replication Instance を作成する

DMS コンソールの［移行または複製］から［プロビジョンドインスタンス］を開き、［プロビジョンドインスタンスを作成］を選択しました。主な設定は以下のとおりです。

作成後、ステータスが Available になることを確認します。

これで、DMS タスクの実行に必要な主要リソースがすべて揃いました。

Aurora MySQL
  └─ dms-cdc-lab-source-aurora

AWS DMS Replication Instance
  └─ dms-cdc-lab-ri

Amazon S3
  └─ dms-cdc-lab-target-s3

ソース・ターゲットエンドポイントの接続をテストする

dms-cdc-lab-ri を指定して、各エンドポイントへ到達できることを確認します。

いずれもステータスが successful になったことを確認しました。

Full Load + CDC を実行する DMS タスクを作成する

DMS コンソールで［移行または複製］の［タスク］を開き、［タスクを作成］を選択しました。主な設定は以下のとおりです。

Full Load + CDC タスクは、以下の順序で処理が進みます。

Full Load
  ↓
キャッシュされた変更の反映
  ↓
CDC による継続レプリケーション

タスク設定はウィザードを使用し、検証に不要な設定はデフォルト値のままとしました。

続いて、移行対象を app.orders に限定するテーブルマッピングを設定します。

同じ内容を JSON で記述すると以下のとおりです。

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "include-orders",
      "object-locator": {
        "schema-name": "app",
        "table-name": "orders"
      },
      "rule-action": "include"
    }
  ]
}

［作成時に自動で開始する］を選択しているため、タスク作成後に Full Load が開始されます。

確認してみた

DMS タスクの Full Load 完了と CDC 実行状態を確認する

DMS コンソールのタスク一覧で、dms-cdc-lab-full-load-cdc の Full Load 進行状況が 100% となり、ステータスが「ロード完了、レプリケーション実行中」になりました。

初期データの Full Load が完了し、以降の変更を CDC として継続的に取得する状態になったことを示します。

S3 に Full Load の Parquet ファイルが出力されていることを確認する

出力先として指定した raw/dms/ 配下を確認すると、app/orders/ に Full Load の Parquet ファイルが出力されていました。

Full Load の対象は、タスク開始前に登録した以下の 2 件です。

Aurora MySQL で INSERT、UPDATE、DELETE を実行する

CDC の出力を確認するため、クエリエディタから app.orders に対して変更操作を実行します。

INSERT INTO app.orders (
  order_id,
  customer_name,
  status,
  amount,
  updated_at
)
VALUES (
  3,
  'Carol',
  'CREATED',
  3000.00,
  NOW()
);

UPDATE app.orders
SET
  status = 'PAID',
  amount = 3300.00,
  updated_at = NOW()
WHERE order_id = 1;

DELETE FROM app.orders
WHERE order_id = 2;

実行後、テーブルの状態を確認します。

SELECT
  order_id,
  customer_name,
  status,
  amount,
  updated_at
FROM app.orders
ORDER BY order_id;

order_id=1 が更新され、order_id=3 が追加され、order_id=2 が削除されていることを確認できました。

S3 に CDC の Parquet ファイルが出力されることを確認する

数分待機してから S3 の出力先を確認すると、CDC 用の Parquet ファイルが追加されていました。

s3://<bucket-name>/raw/dms/app/orders/
└── <CDC の日時パーティション>/
    └── <CDC の Parquet ファイル>

S3 ターゲットエンドポイントで日時パーティションを JST の時間単位で作成するよう設定しているため、CDC データはコミット時刻に対応するプレフィックス配下に出力されます。

Glue Crawler で S3 の Parquet をテーブルとして登録する

S3 に出力された CDC の Parquet ファイルを Athena からクエリするため、AWS Glue Crawler で Glue Data Catalog にテーブル定義を作成します。今回は Full Load と CDC の出力を分けて確認するため、Crawler のデータソースには CDC の日時パーティションを指定します。

s3://<bucket-name>/raw/dms/app/orders/<YYYY>/<MM>/<DD>/<HH>/

Crawler の主な設定は以下のとおりです。

Crawler が使用する IAM ロールには、S3 の CDC 出力先を読み取る権限と、Glue Data Catalog にテーブルを作成・更新する権限を付与します。

• ロール名: AWSGlueServiceRole-dms-cdc-lab-crawler

信頼ポリシー

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "glue.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

S3 読み取り用のインラインポリシーです。<BUCKET_NAME> と日時パーティション部分は、実際の値に置き換えてください。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ListCdcPrefix",
      "Effect": "Allow",
      "Action": [
        "s3:ListBucket",
        "s3:GetBucketLocation"
      ],
      "Resource": "arn:aws:s3:::<BUCKET_NAME>"
    },
    {
      "Sid": "ReadCdcParquetFiles",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject"
      ],
      "Resource": "arn:aws:s3:::<BUCKET_NAME>/raw/dms/app/orders/*"
    }
  ]
}

あわせて、AWS 管理ポリシーの AWSGlueServiceRole をアタッチします。

Crawler を実行し、ステータスが Ready、最終実行結果が Succeeded であることを確認しました。また、Table changes from last run が 1 created となっていることから、Glue Data Catalog にテーブルが 1 件作成されたことを確認できました。

Amazon Athena で CDC データをクエリする

最後に、Glue Crawler が作成したテーブルを Amazon Athena からクエリします。Crawler のデータソースに日時パーティションの末尾（時）まで指定したため、Glue Data Catalog には 18 というテーブル名が作成されました。数値から始まるテーブル名のため、Athena では "18" のように識別子を引用符で囲んで参照します。

まず、クエリ結果の出力先を s3://<bucket-name>/athena-results/ に設定し、テーブルの列定義を確認します。

SHOW COLUMNS IN dms_cdc_lab."18";

操作種別を示す op 列を含め、orders テーブルの各列が登録されていることを確認できました。

続いて、操作種別ごとの件数を確認します。

SELECT
  op,
  COUNT(*) AS event_count
FROM dms_cdc_lab."18"
GROUP BY op
ORDER BY op;

INSERT を示す I、UPDATE を示す U、DELETE を示す D が出力されていることを確認できました。

最後に、CDC レコードの内容を確認します。

SELECT
  op,
  order_id,
  customer_name,
  status,
  amount,
  updated_at
FROM dms_cdc_lab."18"
ORDER BY order_id;

取得できた CDC レコードは以下のとおりです。

order_id=3 の INSERT、order_id=1 の UPDATE、order_id=2 の DELETE が、S3 上の Parquet ファイルに CDC データとして出力されていることを確認できました。

リソースの削除

不要な課金を避けるため、以下の順序で削除します。

1. DMS タスク（dms-cdc-lab-full-load-cdc）を停止してから削除
2. DMS エンドポイント（ソース・ターゲット）を削除
3. DMS Replication Instance（dms-cdc-lab-ri）を削除
4. DMS Replication Subnet Group を削除
5. Aurora クラスター（dms-cdc-lab-aurora）と Writer インスタンスを削除（最終スナップショットは任意）
6. DB サブネットグループ、DB クラスターパラメータグループを削除
7. Glue Data Catalog のテーブル・データベース（dms_cdc_lab）、Glue Crawler を削除
8. S3 バケット内のオブジェクトを削除し、バケットを削除
9. S3 用 Gateway VPC Endpoint、セキュリティグループ、VPC を削除
10. 作成した IAM ロール（dms-cdc-lab-s3-role、AWSGlueServiceRole-dms-cdc-lab-crawler）を削除

まとめ

Aurora MySQL をソースとして、AWS DMS の Full Load + CDC で Amazon S3 に Parquet 形式のデータを出力しました。今回確認できた内容は以下のとおりです。

• Aurora MySQL の既存データを Full Load で S3 に出力できた
• INSERT、UPDATE、DELETE の変更を CDC として S3 に出力できた
• CDC データを JST の時間単位で S3 にパーティション出力できた
• AWS Glue Crawler で CDC の Parquet ファイルをテーブルとして登録できた
• Amazon Athena から I、U、D の操作種別を含む raw CDC データをクエリできた

一方で、Athena で確認したデータは変更イベントを蓄積した raw データです。同一レコードへの複数回の更新や削除イベントを含むため、そのまま最新状態のテーブルとして利用することはできません。
分析用途で最新状態を扱う場合は、後続処理で主キー、操作種別、変更時刻を考慮してデータを統合する必要があります。AWS Glue ETL や Apache Iceberg を利用した curated テーブルの作成は、今後の検証課題とします。
本ブログが誰かの参考となれば幸いです。

参考資料

• AWS Database Migration Service のターゲットとしての Amazon S3 の使用
• MySQL から S3 データレイクへの移行（Step-by-Step Migration）
• MySQL 互換データベースを AWS DMS のソースとして使用する
• レプリケーションインスタンスのためのネットワークのセットアップ
• Aurora のクエリエディタの使用

クラスメソッドオペレーションズ株式会社について

クラスメソッドグループのオペレーション企業です。
運用・保守開発・サポート・情シス・バックオフィスの専門チームが、IT・AI をフル活用した「しくみ」を通じて、お客様の業務代行から課題解決や高付加価値サービスまでを提供するエキスパート集団です。
当社は様々な職種でメンバーを募集しています。
「オペレーション・エクセレンス」と「らしく働く、らしく生きる」を共に実現するカルチャー・しくみ・働き方にご興味がある方は、クラスメソッドオペレーションズ株式会社 コーポレートサイト をぜひご覧ください。※2026 年 1 月にアノテーション株式会社から社名変更しました