AWS DMS:列フィルタリングを避けるべき理由

AWS Database Migration Service (DMS) は、同種・異種のデータベース環境を問わず利用できる、定番のデータベース移行・レプリケーションツールです。Amazon RDS MySQL や Amazon RDS PostgreSQL などをソースとしてデータを移行する際、ソースフィルターを設定すれば、ターゲットデータベースに転送するレコードの数や種類を絞り込めます。ただし、移行をスムーズに進めるためには、フィルターにいくつかの制約があることを理解しておく必要があります。

AWS DMS の概要

AWS DMS は、ソースからターゲットデータベースへ変更を継続的にレプリケートし、ダウンタイムを最小限に抑えてデータ移行を実現します。主な機能は次のとおりです。

• フルロード、CDC、継続的レプリケーション: 初期移行と継続的な同期に対応します。
• スキーマ・テーブルマッピング: オブジェクトを柔軟に選択・変換できます。
• フィルタリング機能: 行レベル・列レベルのフィルタリングで、レプリケート対象データを制御できます。
• マルチデータベースエンジン対応: MySQL、PostgreSQL、SQL Server、Oracle など多様なデータベースに対応します。

もっとも、これらの機能を備えていても、AWS DMS の列フィルタリングは、特に CDC 中の更新処理において期待どおりに動かないことがあります。

AWS DMS の列フィルタリングとは?

AWS DMS の列フィルタリングを使うと、特定の列に対する条件に応じて、レプリケート対象の行を絞り込めます。たとえば「status」列が「active」の行や、「timestamp」列が特定日付より大きい行だけをレプリケートする、といった DMS タスクを設定できます。データの一部だけを移行したい場合や、移行プロセスの中でビジネスロジックを適用したい場合に重宝する機能です。

DMS の列フィルタールールの例を以下に示します。

{
  "rules": [\
    {\
      "rule-type": "selection",\
      "rule-id": "1",\
      "rule-name": "FilterActiveUsers",\
      "object-locator": {\
        "schema-name": "public",\
        "table-name": "users"\
      },\
      "rule-action": "include",\
      "filters": [\
        {\
          "filter-type": "source",\
          "column-name": "status",\
          "filter-conditions": [\
            {\
              "filter-operator": "eq",\
              "value": "active"\
            }\
          ]\
        }\
      ]\
    }\
  ]

ユースケース:データ移行時の列フィルタリング

DMS が列フィルタリングをどう扱うかを検証するため、AWS Database Migration Service (DMS) を使い、ある RDS MySQL 8.0 インスタンスから別の RDS MySQL 8.0 インスタンスへデータを移行する PoC を実施しました。要件はシンプルで、「name」という列が特定の条件 (name = "b") を満たす行だけを、ソーステーブルからターゲットテーブルへレプリケートするというものです。これを実現するため、特定の値で行を絞り込める DMS のソース列フィルタリング機能を使いました。

セットアップ

• ソース:RDS MySQL 8.0.36
• ターゲット:RDS MySQL 8.0.36
• DMS バージョン:Engine 3.5.4
• 移行モード:CDC を伴うフルロード

ソースデータベースに、次のサンプルテーブルを作成しました。

CREATE TABLE `Sample` (
`id` varchar(255) NOT NULL,
`name` varchar(255) NOT NULL,
PRIMARY KEY (`id`)
);

name = 'b' の行だけをレプリケートするよう、列レベルのフィルターを設定した DMS 移行タスクを構成しました。

DMS マッピングルール:

{
"rules": [\
{\
"rule-type": "selection",\
"rule-id": "098947021",\
"rule-name": "098947021",\
"object-locator": {\
"schema-name": "Test",\
"table-name": "Sample"\
},\
"rule-action": "include",\
"filters": [\
{\
"filter-type": "source",\
"column-name": "name",\
"filter-conditions": [\
{\
"filter-operator": "eq",\
"value": "b"\
}\
]\
}\
]\
}\
]
}

ソースフィルター付きのセレクションルール

検証結果と考察:

name = 'b' の行を挿入すると、想定どおりレプリケートされました。name = 'a' の挿入も想定どおり無視されました。一方、更新については挙動に不整合がありました。

• name='a' を name='b' に更新しても、CDC がイベントを捕捉していたにもかかわらず、その更新は無視されました。

DMS の検証機能を有効にしている場合、この挙動によりソースとターゲット間でレコードの不一致が生じ、検証で差分として検出されます。

つまり AWS DMS は更新前の状態に基づいてフィルターを適用するため、列フィルタリングを使うとデータ整合性の問題を招く恐れがあるということです。

根本原因 - AWS DMS によるフィルター適用の仕組み

AWS DMS は、更新が発生する前の行の状態に基づいて列フィルターを評価します。具体的には次のとおりです。

• 挿入イベント: 挿入される行の値に対してフィルターが適用されます。
• 更新イベント: 更新が適用される前の値でフィルターが評価されます。
• 削除イベント: 削除前の行の状態に対してフィルターがチェックされます。

更新は元の行の状態でフィルタリングされるため、name = 'a' から name = 'b' への更新は無視されます。更新前の段階でフィルター条件を満たしていなかったので、その行は一度もレプリケートされません。

この挙動はデータの不整合を引き起こす可能性があります。アプリケーションが更新後の値で行を絞り込むことを前提としていると、本来レプリケートされるはずのデータが届かないという事態が起こり得ます。

回避策:ターゲットデータベース側でフィルタリングする

AWS DMS には更新処理後にフィルターを適用するオプションがないため、次のような回避策が必要になります。

ステップ 1:DMS タスクから列フィルターを外す

フィルタリングを行わず、すべてのデータをレプリケートするよう DMS タスクを修正します。

ステップ 2:ターゲットデータベース側にフィルタリングを実装する

ターゲットデータベースの Sample テーブルへ直接レプリケートするのではなく、いったんステージングテーブルにレプリケートし、トリガーやストアドプロシージャを使って有効な行だけを Sample テーブルへ移します。

まず、ターゲットの RDS MySQL データベースにステージングテーブル (sample_staging) を作成します。このテーブルは、Sample テーブルに移す前のレプリケートデータを一時的に保持する役割を担います。

MySQL の場合の例は次のとおりです。

CREATE TABLE sample_staging (
id INT PRIMARY KEY,
name VARCHAR(255)
);

ステップ 3:ストアドプロシージャを作成する

ステージングテーブル (sample_staging) からターゲットテーブル (Sample) へ、有効な行 (例:name = 'b' の行) を移すストアドプロシージャを作成します。このプロシージャは挿入と更新の両方を処理します。

CREATE PROCEDURE MoveValidRowsToSample()
BEGIN
-- Insert or update valid rows in the Sample table
INSERT INTO Sample (id, name)
SELECT id, name
FROM sample_staging
WHERE name = 'b'
ON DUPLICATE KEY UPDATE
name = VALUES(name);
--Delete valid rows from the staging table
DELETE FROM sample_staging
WHERE name = 'b';
END

ステップ 4:トリガーを作成する

ステージングテーブル (sample_staging) で行が挿入または更新されるたびに、自動でストアドプロシージャを呼び出すトリガーを作成します。

CREATE TRIGGER after_insert_sample_staging
AFTER INSERT ON sample_staging
FOR EACH ROW
BEGIN
CALL MoveValidRowsToSample();
END

CREATE TRIGGER after_update_sample_staging
AFTER UPDATE ON sample_staging
FOR EACH ROW
BEGIN
CALL MoveValidRowsToSample();
END

処理の流れ

• ソースデータベース:元データを保持します。
• AWS DMS:ソースから、ターゲットの RDS MySQL データベースの sample_staging テーブルへデータをレプリケートします。
• ステージングテーブル (sample_staging):レプリケートされたデータをすべて一時的に保持します。
• トリガー:sample_staging テーブルで挿入や更新が発生するたびに、自動でストアドプロシージャを呼び出します。
• ストアドプロシージャ:有効な行 (name = 'b' の行) を sample_staging から Sample テーブルへ移します。
• ターゲットテーブル (Sample):トリガーとストアドプロシージャによる処理を経て、有効な行だけが残ります。

PostgreSQL の場合は、BEFORE INSERT OR UPDATE トリガーを使って同様の仕組みを実装できます。

RDS PostgreSQL での検証

この挙動が MySQL 固有のものではないかを確認するため、RDS PostgreSQL でも同じテストを行いました。結果はまったく同じでした。

• 更新によってフィルター条件を満たすようになった行は、ターゲットテーブルにレプリケートされませんでした。
• これにより、本問題はデータベース固有のものではなく、CDC 中における DMS の列フィルター処理方式に起因する制約であることが裏付けられました。