はじめに
今回はこのアップデートを確認してみます！
Amazon Aurora DSQL が変更データキャプチャ（CDC）をプレビューでサポートしました。データベース上の INSERT / UPDATE / DELETE 操作を検知し、Amazon Kinesis Data Streams にストリーミングしてくれます。
これまでもリレーショナルデータベースの変更をきっかけに後続処理を動かす仕組みはありました。例えば Aurora MySQL ではトリガーから AWS Lambda を直接呼び出す統合機能があります。
Aurora DSQL の変更データキャプチャはそれとはアプローチが異なり、変更イベントを Kinesis Data Streams にストリーミングする方式です。フルマネージドかつデータベースワークロードへの影響ゼロで配信できるのが特徴ですね。
詳細は AWS 公式ブログに書いてあるので、あわせて参照ください。
この記事で学べること
- Aurora DSQL の変更データキャプチャのセットアップ手順（AWS CDK + マネジメントコンソール）
- 変更データキャプチャイベントの構造と各操作（INSERT / UPDATE / DELETE）での違い
- Lambda で変更データキャプチャイベントを受け取って処理する方法
- プレビュー段階の制約事項
前提知識・条件
- 本機能はプレビュー段階です。GA までに仕様が大きく変更される可能性があります
- Aurora DSQL の基本的な概要を理解していること
やってみた
では、簡単に動作確認していきます。
事前にテーブルとデータを準備しておき、その後に変更データキャプチャを設定し、データの更新・削除を実施してみます。
全体の構成としては、Aurora DSQL のテーブル変更が変更データキャプチャによって Kinesis Data Streams に流れます。
Lambda がそのイベントを受け取ってログに出力するというシンプルなパイプラインです。
Aurora DSQL（orders テーブル）
↓ 変更データキャプチャでイベント自動キャプチャ
Kinesis Data Streams
↓ イベントソースマッピング
Lambda（ログ出力）
0.準備
今回新機能の変更データキャプチャ以外のところは CDK で作ってしまいます。
CDK で作成するリソースは以下です。
- Aurora DSQL クラスタ
- Kinesis Data Streams
- Lambda 関数（変更データキャプチャイベントをログ出力するコンシューマ）
- Lambda と Kinesis のイベントソースマッピング
CDK のスタックは以下の通りです。
Lambda のコードはシンプルに変更データキャプチャイベントをパースしてログに出力するだけのものです。Kinesis から受け取るレコードの data は Base64 エンコードされているので、デコードしてから JSON パースしています。
また以下の DDL で簡単なテーブルを作っておきます。
CREATE TABLE IF NOT EXISTS orders (
id INT PRIMARY KEY,
customer_name TEXT NOT NULL,
item TEXT NOT NULL,
quantity INT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO orders (id, customer_name, item, quantity, status) VALUES
(1, '田中太郎', 'ノートPC', 1, 'pending'),
(2, '鈴木花子', 'マウス', 2, 'pending'),
(3, '佐藤次郎', 'キーボード', 1, 'pending');
1. 変更ストリームの指定
テーブルとデータの準備ができたら、マネジメントコンソールから変更ストリームを作成してみます。
Aurora DSQL のコンソールでクラスタを選択すると、新しいタブが増えていますね！
ここから名前や、送信先の Kinesis データストリームを選択します。
変更ストリームのステータスが ACTIVE になれば準備完了です。ここからデータを変更して、イベントが流れてくるか確認していきます。
1. データ更新（update）
orders テーブルの id=1 のレコードのステータスを shipped に更新してみます。
UPDATE orders SET status = 'shipped' WHERE id = 1;
しばらくすると Lambda の Amazon CloudWatch Logs に以下のようなイベントが出力されました。
{
"operation": "INSERT/UPDATE",
"table": "public.orders",
"txId": "7fty5hq4ezu7jlhcfn2uehlspi",
"type": "full",
"before": null,
"after": {
"id": 1,
"customer_name": "田中太郎",
"item": "ノートPC",
"quantity": 1,
"status": "shipped",
"created_at": 1779096822929852
},
"timestamp": 1779097432256459300
}
いくつか気になるポイントがあります。
まず operation が INSERT/UPDATE になっています。これはプレビュー段階の仕様で、INSERT と UPDATE が区別されず同じ op: "c" として出力されます。
ドキュメントによると、GA までに UPDATE 用の op: "u" が追加される予定とのことです。
Before general availability, we will add new operation types ("op": "u" for updates) to your stream payload. To ensure your application handles these changes without modification, treat any unrecognized op value as an upsert by applying the after payload.
次に before が null です。UPDATE であっても変更前の値は含まれず、after に全カラムの最新値が入ります。行全体が入ってくるので、コンシューマ側で「何が変わったか」を知りたい場合は前回の状態を自前で保持しておく必要があります。
created_at がマイクロ秒のエポック値（1779096822929852）で返ってきているのも特徴的ですね。TIMESTAMP 型はこの形式で配信されるようです。
2. データ削除（delete）
続いて id=3 のレコードを削除してみます。
DELETE FROM orders WHERE id = 3;
Lambda のログに以下のイベントが出力されました。
{
"operation": "DELETE",
"table": "public.orders",
"txId": "5vty5hvnzequ7yztjhw47nj4p4",
"type": "full",
"before": {
"id": 3
},
"after": null,
"timestamp": 1779097581459401000
}
DELETE の場合は UPDATE とは異なり、op: "d" としてきちんと区別されています。before には主キーカラム（id）のみが含まれ、after は null です。
ここで before フィールドの役割を整理しておくと、このフィールドは「変更前の行データ」を保持するものではなく、DELETE 時に「どの行が消されたか」を特定するためだけに使われます。
まとめ
今回は Aurora DSQL の変更データキャプチャ（CDC）プレビューを試してみました。
CDK でクラスタ・Kinesis・Lambda を構築し、マネジメントコンソールから変更ストリームを作成しました。これだけでデータベースの変更イベントをリアルタイムに受け取れることが確認できました。
確認できた変更データキャプチャイベントの特徴をまとめます。
| 操作 | op | before | after |
|---|---|---|---|
| INSERT | "c" | null | 全カラム |
| UPDATE | "c"（プレビュー段階） | null | 全カラム（最新値） |
| DELETE | "d" | 主キーのみ | null |
なお、プレビュー段階の制約として以下の点があります。
- INSERT と UPDATE が区別されない（両方
op: "c"）。GA でop: "u"が追加予定 - クラスタレベルで全テーブルの変更がキャプチャされる（テーブル単位のフィルタリングは未サポート）
- at-least-once delivery のため、コンシューマ側での重複排除が必要
この機能により、DSQLをトリガーとしたイベント駆動のアーキテクチャが設計しやすくなったかなと感じました！
現時点ではクラスタ内の全テーブルの変更が流れてくるため、特定テーブルだけ必要な場合はコンシューマ側でフィルタリングする必要があります。今後テーブル単位の選択がサポートされると、より実用的になりそうですね。