📊

Topological SortでBigQueryのジョブをスケジューリングする

2023/12/14に公開

Python

概要

依存関係を持つBigQueryのSQLのジョブを、 Topological Sortによってスケジューリングする方法を紹介します。
dbtやDataformを利用すればおのずと依存関係に基づいたデータパイプラインを構築できますが、そのようなツールを導入しない簡易的な依存関係の解決手段としてお見知りおきください。

サンプル

サンプルのリポジトリです。今回はgraphlibを利用するためPythonで実装しています。

ユースケースとして、ECサイトで「ユーザーがお気に入りに登録しているショップで、20%以上の値引きがあり、なおかつ在庫が存在する商品」を取得するためのジョブを想定します。

スキーマ

データソースとなるテーブルのスキーマです。

schema

SQL

5つのSQLを実行し、最後にfavorite_discount_goodsテーブルを作成します。

schema

解説

主な処理の流れです。

1. SQLファイルの取得

はじめにSQLファイルを取得し、SQL classのinstanceを生成します。
SQLファイルはsqlsディレクトリ下のすべての.sqlファイルが対象です。

SQL classはread()でファイルを読み込み、get_dependencies()で自身の作成するテーブル名と、依存するテーブル名のlistからなるdictionaryを返します。

例えば、sqls/reporting/favorite_discount_goods.sql はこのような値を返します。

{'project.reporting.favorite_discount_goods': ['project.staging.discount_goods',
                                               'project.staging.favorite_shops']}

2. 依存関係のグラフへの変換

すべてのSQLの依存関係からグラフを作成すると、以下の結果になります。

{'project.reporting.favorite_discount_goods': ['project.staging.discount_goods',
                                               'project.staging.favorite_shops'],
 'project.staging.discount_goods': ['project.sources.discounts',
                                    'project.staging.shop_goods_with_stocks'],
 'project.staging.favorite_shops': ['project.sources.favorite_shops',
                                    'project.sources.shops',
                                    'project.sources.users'],
 'project.staging.shop_goods': ['project.sources.goods',
                                'project.sources.shop_goods',
                                'project.sources.shops'],
 'project.staging.shop_goods_with_stocks': ['project.sources.stocks',
                                            'project.staging.shop_goods']}

3. ジョブのスケジュールと実行

作成したグラフにTopological Sortをかけます。

結果、job_sequencesにはネストしたテーブル名のlistがセットされます。

[['project.sources.stocks',
  'project.sources.discounts',
  'project.sources.favorite_shops',
  'project.sources.shops',
  'project.sources.users',
  'project.sources.goods',
  'project.sources.shop_goods'],
 ['project.staging.favorite_shops',
  'project.staging.shop_goods'],
 ['project.staging.shop_goods_with_stocks'],
 ['project.staging.discount_goods'],
 ['project.reporting.favorite_discount_goods']]

このように依存関係にもとづいた順番になっていることがわかります。
先頭のlistはデータソースのテーブルのため、sliceで削除しています。

あとはlistごとに並列でジョブを実行します。

futuresを出力すると、以下のような結果になります。

{<Future at 0x10477b800 state=finished returned NoneType>: 'project.staging.favorite_shops',
 <Future at 0x10479c5f0 state=finished returned NoneType>: 'project.staging.shop_goods'}
{<Future at 0x1040143b0 state=finished returned NoneType>: 'project.staging.shop_goods_with_stocks'}
{<Future at 0x104288560 state=finished returned NoneType>: 'project.staging.discount_goods'}
{<Future at 0x10477b6b0 state=finished returned NoneType>: 'project.reporting.favorite_discount_goods'}

4. グラフの描画

おまけで、最後にDigraphでグラフをPDFファイルに出力しています。
冒頭のSQLのグラフはこちらで出力したものです。

まとめ

Topological SortでBigQueryのジョブをスケジューリングする方法を紹介しました。