データウェアハウス移行のために何をしたか？
データウェアハウスの移行にあたり、まず実施されたのは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テクノロジーズ
