【データ基盤構築/AWS】DMSとSnowpipeを使ってRDSのデータをSnowflakeにロードする時につまずいたこと

今回の課題

DMS(Database Migration Service)とSnowpipeを使うことで、
RDS上の変更をほぼリアルタイムでデータウェアハウスに転送することが可能だということを知ったため試してみることにした。

実装手順

基本的に、下記の記事の通りに進めたので、
本記事での解説は割愛して、実装中につまずいたところを中心に書いていきたいと思います。

実装した手順を簡潔に書くと、以下の流れになります。

1. DMSのレプリケーションインスタンスの設定。
2. データ転送元のRDSに対して、ソースエンドポイントを作成。
3. データ転送先のS3に対して、ターゲットエンドポイントを作成。
4. データベース移行タスクを作成。
   ※データベース移行タスクの作成では、移行タイプを既存のデータを移行して、継続的な変更をレプリケートするまたはデータ変更のみをレプリケートするを選択し、RDSの変更が継続的にS3にロードされるように設定します。
5. Snowpipeを作成。

つまずいたポイント

RDSのデータに変更があった際に、変更を検知してS3にデータをロードしたいので、
移行タイプを既存のデータを移行して、継続的な変更をレプリケートするまたは、データ変更のみをレプリケートするに設定していたのだが、
そうすると、データベース移行タスクを実行しても、下記の画像のように失敗に終わってしまう。

色々試行錯誤してみたところ、
データ転送元のRDSに対してソースエンドポイントを作成する段階で、以下2点の設定ができていなかったことが原因だと分かった。

できていなかった設定について

RDSのMYSQLバイナリログ保持時間を変更

まず1つ目のできていなかった設定は、バイナリログの保持時間の変更。
デフォルトだと、バイナリログを保持しない状態となっている。

以下の記事のように、
バイナリログの保持時間についての値を_NULL(バイナリログを保持しない状態)から、24 (24時間保持する状態)に変更する設定を行う必要があった。

このように設定することによって、レプリケーション時に参照するバイナリログがすぐに消えてしまわないように調整することができた。

※バイナリログとは
MYSQLサーバーインスタンスで行われたデータ変更に関する情報を含む、一連のログファイルのこと。

RDS Parameter Groupを変更する

2つ目のできていなかった設定は、RDS Parameter Groupの変更。
RDS Parameter Groupとは、データベースの設定をテンプレートとして管理できるような機能のこと。

以下の手順で、DBパラメータグループの作成・編集して、現在利用しているRDSの設定に使用する必要があった。

1)パラメータグループを新規作成する

RDSのコンソール画面からパラメータグループの作成を行う画面に飛ぶ。

ここで、現在使用しているバージョンと同じ(今回はmysql8.0)パラメタグループを選択し、パラメータグループの新規作成を行う。

2)パラメータの編集を行う

DBパラメータグループを作ったら、下記の作成したDBパラメータグループのページから、
パラメータの編集ページに飛ぶ。

そして、こちらのドキュメントのように、1)で作成したDBパラメータグループの2つのパタメータを下記のように変更する。

• binlog_format = ROW
• binlog_checksum = NONE

3)現在利用しているRDSのDBパラメータグループを変更する

RDSのDBインスタンス変更画面の、追加設定のデータベースの選択肢にて、
DBパラメータグループを2)で作成したものに変更する。

これによって、RDSの設定を変えることができた。

4)RDSインスタンスを再起動する。

最後にRDSを再起動する。
これによって、3)で行なったRDSの設定の変更を反映させることができる。

まとめ

以上のように進めることで、
データベース移行タスクが失敗に終わってしまう不具合を解決することができた！

そしてそれによって、
RDSに変更があった時にはDMSがS3に変更データをロードし、
SnowpipeがSnowflakeのテーブルにほぼリアルタイムでデータをロードする。
という機能を実装することができた。

MySQLは「バイナリログ」を使って、データベースの操作履歴を記録しているため、「バイナリログ」周りの設定を少し調整する必要があったようだった。

参考記事