こんにちは、MLデータ部データ基盤ブロックの奥山（@pokoyakazan）です。趣味の範疇ですが、「ぽこやかざん」という名前でラジオ投稿や大喜利の大会に出たり、「下町モルモット」というコンビで週末に漫才をしたりしています。私は普段、全社データ基盤の開発・運用を担当しており、このデータ基盤はGCPのBigQuery上に構築されています。そして、データ基盤内の各テーブルは、大きく分けて以下の2種類に分類されます。
本記事では、後者のデータマートを集計するジョブを制御するワークフローエンジンを、DigdagからCloud Composerに移行した事例について紹介します。Cloud Composerとは、GCPにてApache Airflowをマネージドに提供するサービスです。
なお、本記事では、Cloud Composer・Apache Airflowそれぞれのバージョンは以下のものとして話を進めます。
そのため、記事内で参考情報リンクとして貼っている公式ドキュメントについても、こちらのバージョンのものとなります。
以下の記事でもご紹介した通り、マート集計処理の実体はデータ基盤利用者が作成したSQLファイルで、全てGitHubで管理されています。
SQLファイルにはSELECT文のみが記述されており、UPDATEやDELETEといったDMLは記載されていません。
あるマートが他のマートを参照している（依存関係がある）場合、集計の順番を間違えるとデータに不整合が発生してしまいます。例えば、existing_table1, existing_table2という一次テーブルが存在するとし、以下のような集計クエリを持つ5つのマートを構築したい場合を考えます。
table1.sql
/assets/images/7661308/original/3a023d97-0ee4-488d-a826-b9ee662c0ee1?1632994701)
SELECT * FROM `project.dataset.existing_table1`;table2.sql
SELECT * FROM `project.dataset.existing_table2`;table3.sql
SELECT * FROM `project.dataset.table2`;table4.sql
SELECT * FROM `project.dataset.table1`
UNION ALL
SELECT * FROM `project.dataset.table3`;table5.sql
SELECT * FROM `project.dataset.table3`;この場合、「table3の前にtable2」「table4の前にtable1とtable3」「table5の前にtable3」が集計されている必要があります。
移行前のシステムのデータマート集計方法
マート間の依存関係は、各マートのSQLファイル内の、FROMもしくはJOINの直後にくるマート（自己参照は除く）を調べるとわかります。FROM, JOINの後ろに書かれているマートは、SQLを実行するマートよりも前に集計しなければなりません。そのため、「FROM, JOINの後ろのマート」→「SQLを実行するマート」というように依存関係グラフを作成していきます。例えば、上記の5つのSQLからマート間の依存関係グラフを作成すると以下のようになります。
これをPythonコードで実装していきます。まず、各マートのSQLファイルから以下の正規表現を使って参照先となるマートを抽出します。
(?i)(?<=FROM|JOIN)[\s \n]*`(.+?)`そして、参照元: 参照先という形のDictを作成します。
{
'table1': [], # table1の依存先
'table2': [], # table2の依存先
'table3': ['table2'], # table3の依存先
'table4': ['table1', 'table3'], # table4の依存先
'table5': ['table3'] # table5の依存先
}作成した依存関係グラフを利用し、マートの集計順序を担保したまま、可能な限り処理を並列化していきます。具体的には、並列実行しても問題のないマート同士をグループ化します。まず、親ノードがないマート群をリストに追加し、追加したマートをグラフから削除します。そして、もう一度親ノードがないマート群をリストに追加し、追加したマートをグラフから削除…というのを繰り返していきます。
結果として、以下のような集計グループのリストができあがります。
[[table1, table2], [table3], [table4, table5]]各集計グループ（マートのリスト）は、リストの先頭から順番に実行可能で、同じ集計グループ内のマートの集計は並列化できます。結果として、集計の流れは以下のようになります。
データマート集計ジョブの課題
上記の方法で集計すると、依存関係に沿った集計順序が担保され、同じ集計グループ内では処理を並列化できます。ただし、この方法にはいくつか課題も存在します。
1つのマートの集計が失敗すると後続のグループに属する全てのマートの集計が停止する
あるマートの集計ジョブがエラー終了した場合、このマートと同じ集計グループに属するマートについては、処理が並列化されているため影響を受けません。しかし、失敗したマートが属する集計グループより後のグループは、全て処理が停止してしまいます。例えば、table1の集計がこけた場合、以下のようになります。
集計グループ単位で見ると、先頭の集計グループの処理が失敗しているので、2番目と3番目の集計グループの処理は開始されません。そのため、table1に依存しないtable[3, 5]の集計は実行されてほしいところですが、これらのマートの集計も停止してしまいます。
の更新が遅延している場合、以下のようになります。データマートは、一次テーブルから必要なデータを使いやすい形に加工し抽出したテーブルです。そのため、一次テーブルが更新される前に、一次テーブルを参照しているマートの集計が行われるとデータの不整合が発生します。そこで、一次テーブルを参照するマートは、一次テーブルが正常に更新されるまで集計開始を待つ必要があります。さらに、集計グループ内の1つのマートのみ集計を停止させることはできないため、その場合は集計グループ自体の実行を停止（グループ内の全マートの集計を停止）させる必要があります。つまり、なんらかの理由で一次テーブルの更新が失敗・遅延すると、この一次テーブルを参照するマートが属する集計グループ内全てのマートの集計タスクが実行されません。また、一次テーブルに依存するマートの集計タスクは先頭の集計グループに属することが多いため、一次テーブルの更新遅延はマート集計ジョブ全体の大幅な遅延に繋がります。例えば、一次テーブルexisting_table1の更新が遅延している場合、以下のようになります。
続きはこちら
/assets/images/7661308/original/3a023d97-0ee4-488d-a826-b9ee662c0ee1?1632994701)
/assets/images/7661308/original/3a023d97-0ee4-488d-a826-b9ee662c0ee1?1632994701)
