Amazon AuroraとSnowflakeをDynamic Tableを用いてリアルタイム連携したお話
この記事はFinatextグループ10周年記念アドベントカレンダー5日目の記事です。
昨日は齊藤さんが「愛される会社であり続けるために。人気漫画から見えるヒント」という記事を公開しています。
はじめに
ナウキャストでデータエンジニアをしている翁長と申します。
現在、クレジットカードの売上データに対し店舗・ブランド・企業名・業種名など様々なセグメントで分析するためのマッピングを実現するチームにてお仕事をしてます。
マッピングとは
少しばかりナウキャストのマッピングチームの業務について紹介させてください。私達マッピングチームとしての業務は自動化部分とオペレーターの関わる人力部分の２種類が存在します。
一次処理に該当する自動化部分については、あるタグを作成したキーワードに基づいて自動で付与する処理や、店舗名に対しブランドや業種名などを付与する処理を日次でAirflowを用いて実行するマッピング処理となります。こちらは完全にエンジニア側が構築・運用するシステムです。
もう一部は店舗名をもとに業種やブランドなどを調査し、タグの判定や値、キーワードの策定までを含めて行う人力作業となります。こちらはマッピング調査などで培った知見を元にオペレーターと呼ばれる方々が策定してく業務となっており、この策定されたブランドや業種・キーワードの有効性チェックをアドホックに実行したり、データベースに反映するシステムを開発・運用するのがエンジニア側の任務となっています。
移行の背景
現在マッピングチームの用いているデータではRDSに売上データに対するマッピングの結果を保存・活用しています。また、データ同士の結びつきは莫大で様々な処理を逐一RDSからデータを取得しPythonで加工処理と、RDSとアプリ処理側双方に負担がかかり、正直処理が早いとは言えない環境となってました。
現在はこのデータの処理部分をSnowflakeのマシンパワーに期待し、マッピング処理の高速化を実現している過程となっています。
基盤の紹介
現在のマッピングの基盤は個人情報が含まれていない環境をcentral, 含まれているデータを扱う際はsecure環境と、Snowflakeの環境を別途分けて構築・運用しています。
ナウキャストのマッピングチームが扱うクレジットカードの売上情報には一部個人情報に該当するデータが入っているため、パブリックな環境に出しては行けない情報があります。その情報セキュリティの担保を実現したセキュアな環境を実現するため、SnowflakeにはPrivateLinkを使用した接続を実現しています。
また、3次利用として情報セキュリティを担保しながらEC2やWorkspacesからのSnowflakeへの接続を実現したり、Streamlit in Snowflakeをデータ可視化に利用したりとマッピングチーム内では積極的に活用を進めています。
問題点
今回のチャレンジ要素となっていたDynamic Tableを用いたリアルタイム連携実装の前には、一日一回のRDSフルバックアップを用いたバルク更新でSnowflakeのデータ更新を構築していました。
(日次の更新を実行している基盤については、DSPUに所属するKevinさんが以前に公開した記事がございます、ぜひとも御覧ください！）
日次更新でもRDSに比べると見違えるような処理速度の上昇が達成できましたが、前述した通りマッピングの作業結果を保存するリソース部分はAWS環境内のRDSとなっており、すぐにでも作業結果をSnowflake側に反映し、他の作業との整合性の担保やデータの取得を必要とする場面が生まれてきます。そのためにAWSのRDS側にデータの変化が起きた際に、Snowflake側にできるだけ早くデータを同期させるシステムの構築が求められたというのが始まりです。
解決の道筋
リアルタイム連携の実現方法ですが、他のテックブログに偉大な先輩方がAmazon RDS (Amazon Aurora)-> Snowflakeのリアルタイム連携の実現方法を執筆して共有しております。
しかし、大方2021年の記事が最新となっておりSnowflakeのサポートにリアルタイム連携を実現させる何かしらの新しいアーキテクチャやコネクタ等は実装されたか確認したところ、2023年7月頃の返答としては「特に用意がない」とのことでした。
返答を踏まえ、同等のリソースで要件を実現しようと試みたところSnowflake側に新しくDynamic Tableがpublicになったことをウェビナー等で拝聴し、私共がパイプラインの管理で使用しているdbtで構築を実現できることから、チャレンジ要素として基盤に入れ込むこととしました。
AWS DMSについて
AWS DMSではデータソースとしてAmazon Aurora MySQLに接続し、バイナリログを用いて変更データキャプチャ（CDC）を取得しています。このCDCデータにはデータ内容の他にDMLの判断に使えるColumnやCDCとしての取り込み時刻など、後述する増分更新に使える情報が付与されており、Snowflake側へ取り込み後は容易にデータを活用することが可能となっていました。
また、接続情報についてはKMS、SSLとその証明書（DMS専用で別途取り込む必要あり）、DB接続用のSecrets Manager、IAM Roleとセキュリティ要件を満たした接続方法を提供してくれています。
構築部分ではマッピングチームではネットに上がっているこれまでの事例や費用の概算を鑑み、サーバレスではなくインスタンスでDMSを運用することにしました。
当初インスタンスサイズ的にはd3.smallで問題なく捌けていたのですが、5万件を超えるデータの変更が起きた際にメモリ不足エラーでタスクが停止する事象が発生しだしたため、現在は余裕を持ってd3.largeで運用中しています。
運用時に非常にありがたく感じた点は、Database migration tasksのMapping rulesで柔軟にschemaやdatabaseのcolumnなどの取り込みを柔軟に指定・変更できる点や、他のAWSアカウントのS3へのPUT, DELETEも容易に実現できる点。
躓いたポイントとしては、参照するインスタンスがPrivate IPしか持たない場合、特定のバージョンより特定のリソースに対してPrivatelink経由でしかアクセスできない点などがありました。
また、AWS DMSは結構に停止するという記事をいくつか見かけましたが、初期のリソース不足で発生したエラー以外は一度も完全停止したことがないので安心＆拍子抜けというのが現状です。
運用についてはDMSのevent subscriptionも用意があるため、簡単にtaskの状況をSlack等に連携できました。
Snowflake側について
AWS DMSを用いてchange data captureのデータをSnowflakeに連携することができたので、このままでも問題なくリアルタイム連携は実現可能です。しかしながら、このCDCのPipeline構築前に作成した日次更新のExternal TableがSnowflake上にすでに存在するので、こちらを利用しバックアップからの増分更新を実現することにしました。
このExternal Tableで日次処理時のリカバリが可能なので、万一AWS DMSのタスクが停止しCDCの更新が滞ったとしても復旧の工数が最小限で済むこともメリットだと考えてます。
Dynamic TableはStream, Taskを一本化できる新しいTableであり、管理が容易になるというメリットがあります。ただ、構築時にいくつかの縛りや考慮点があったため、その点を記載します。
- 参照先のテーブルとしてExternal Table, Viewは不可
- 参照するテーブルを再作成した際、一部のデータ量が多いテーブルのみ更新エラーが発生した
予測ではありますがDynamic Tableの仕様でTime Travelを利用している節があるので、参照先のテーブルを永続テーブルとしないとエラーとなるようです。また、dbtで参照するTableを再作成した際に何故かデータ量の多いテーブルのみエラーが頻発したため、中身の洗い替えのみを行いテーブルの再作成を避けることで解決しました。
（Snowflakeのサポート様本当にお世話になりました。）
以下はdbtのpre_hookを用い、先に洗い替えを実行している例になります。
{{
config(
materialized="incremental",
pre_hook="""
{% if is_incremental() %}
TRUNCATE TABLE {{ this }}
{% endif %}
""",
)
}}
select ...まとめ
public公開当初は参照テーブル起因でエラーが頻発したり、他アカウントへの共有が直には無理だったり、SISで参照不可だったりと色々物足りない感のあったDynamic Tableですが、現状はそれらも解決し非常に簡単に自動更新されるTableを作成できるようになりました。皆様もぜひDynamic Tableを触ってデータ基盤に活用して頂ければと願います。
おわりに
ナウキャストでは一緒に働く仲間を募集中です。少しでも興味を持たれた方は採用デックを是非ご覧ください！
明日は大木さんによる「AWS re:Invent 2023に参加してきました」という記事です。お楽しみに！
