運用負荷軽減のため、よりマネージドなBigQueryへ移行
かつてZOZOTOWNのインフラでは、データウェアハウスとしてRedshiftが用いられていた。当時のデータフローは下図の通りだ。オンプレミスで動いているZOZOTOWNの各種サービスとAWSのクラウド環境をAWS Direct Connectで結び、ETLツールのTalendによってAmazon S3（以下、S3）にCSVファイルをコピーしていた。
配置されたCSVファイルは、Amazon Data Pipelineを経由してRedshiftにコピーされる。その後、Redshift上で集計用のクエリが実行されるという流れになっていた。
S3には、ZOZOTOWNやWEAR（ZOZOテクノロジーズが開発・運用する、日本最大級のファッションコーディネートアプリ）のマスタデータ、Google Analytics 360から取得したWeb・ネイティブアプリのアクセスログ、メールやプッシュ通知の配信ログなどが格納されている。総テーブル数は100を超え、データサイズは1TBを超えるという。
「RedshiftからBigQueryへの移行を決めた主な理由は、運用負荷を軽減するためです。社内にRedshiftの知見を持つメンバーがそれほど多くなかったため、データ量が今後も増え続けた場合、自分たちでデータベースを運用していく自信がありませんでした。BigQueryの方がよりマネージドなサービスであるため、運用の負担を減らせると考えたのです」
BigQueryにはインデックスという概念がなく、マシンパワーを使って強引にフルスキャンを走らせる仕組みになっている。瞬間的には100台以上のノードで計算されることもあるという。ストレージの量は実質無制限で、いくらでも使うことができる。データの保存のための価格はS3と同程度か少し安いくらいだ。
Amazon Athenaも候補にあがったが、使用技術を検討していた当時はまだ登場したばかりだったことや、クエリの実行速度はBigQueryの方が高速であることから、BigQueryを採用することとなった。
データウェアハウス移行のために何をしたか？
データウェアハウスの移行にあたり、まず実施されたのはS3にあるDataLakeのCSVファイルをBigQueryに日次でロードする処理の実装だ。用いられた要素技術は、TreasureData製のETLツールEmbulkとワークフロー管理ツールDigdagである。
Embulkは大量のデータをバッチ転送することに特化しており、データの入力・加工・出力をする部分はプラグインとして提供されている。設定ファイルはYAMLによる記述だ。Digdagは大量のEmbulkジョブを効率的に管理でき、並列実行・リトライ・タスクの実行状況の可視化と、実行ログの表示などを行える。こちらも、YAML相当のフォーマットで設定ファイルを記述する。
データ転送やワークフロー管理のいずれの機能もTalendにあった。なぜ、Talendを使い続けるのではなく、あえてEmbulkとDigdagを用いたのだろうか。
「理由は大きく2つあります。1つ目は、『普通』のソフトウェア開発のノウハウを生かしたかったこと。例えば、転送設定のワークフローのYAMLファイルをGitHubでバージョン管理する、本番環境に出す前にチーム内でレビューする、CircleCIなどのCI SaaSで自動テストを行う、本番反映作業を自動化するといったことを実現したかったのです。
2つ目は、Digdagを使うことでタスクの並列度の調整を容易にしたかったこと。Embulkは可能な限りマルチスレッドで動作し、マシンのCPUを効率的に使い切ります。高性能なCPUを持つインスタンスを増やして複数台で分散実行させることで、テーブルの転送時間を短縮したかったのです」
データの移行手順は「【1】S3の全データcsvファイルに対してembulk guessコマンドを実行し、転送設定の雛形を作る」「【2】すべてのファイルをBigQueryに転送する」「【3】RedshiftとBigQueryのデータ差分がなくなるまで、転送設定を修正し続ける」という3つのステップに分かれている。
なぜ、移行にあたりデータ差分が生じてしまうのだろうか。その原因の多くは、Embulkの設定ファイルにおける型指定ミスだ。embulk guessコマンドを用いた場合、CSVファイルのスキーマを100％の精度で推測することはできない。これはEmbulkのバグではなく、CSVファイルが型情報を持たないことに起因しているものだ。
検証を続けるなかで、Embulkのcsvパーサーが文字列中の'\0'を正しくパースできないという不具合にも苦しんだという。これらの課題を地道に解決しながら、転送設定の修正作業を続けていった。
また、旧アーキテクチャではRedshiftCopyActivityのinsertModeオプションを用いることで、APPEND（テーブルに追記）やOVERWRITE_EXISTING（主キーを指定し、主キーの重複があったら上書き）などのテーブル差分更新を実現していた。だが、BigQueryには同じ機能が備わっていないため、同等の差分更新処理を自分たちの手で作る必要があった。
APPENDはSDKからクエリを投げるときのwriteオプションでappendを指定することによって、OVERWRITE_EXISTINGはWINDOW関数で旧テーブルのデータか新テーブルのデータかを判別し、重複する場合は新テーブルのデータのみを残す対応を行うことで実現していった。これらに加えて、トランザクションの存在しないBigQueryで冪等な差分更新を実現するための仕組みも新規作成したという。
さらに、データマートの移行も行われた。これは、RedshiftのSQL（PostgreSQL互換のSQL）で書かれたデータマート集計用のクエリを、BigQueryのSQLに変換する作業である。
両データベースで使用できるSQLの関数や挙動には差異があるため、テストをしては地道にSQLを修正する作業が続いた。その他にも、データマートの集計処理をより適切なものにするため、数多くの工夫がなされたという。
最終的に、移行後のデータフローは上図のようになった。今後もより良いアーキテクチャを目指すため、さらなる更新を予定していると、塩崎氏は話す。
「今後の展望として、『オンプレミスのDB上にある集計処理をBigQueryへオフロードする』『Talendで行っている転送処理をEmbulkに置き換える』『全体のリードタイムを減らすためにS3を介さずにオンプレミスのDBから直接BigQueryにデータを送る』などを実現したいと考えています」
お問い合わせ
株式会社ZOZOテクノロジーズ
