dbtは「ETL (ELT)」の「T」、すなわち変換を扱うためのツールだが、dbtのPythonモデルを活用することで、ExtractやLoadも実行できる。抽出からロード、そして変換までの一連の流れをdbtだけで実行することで、シンプルな実装を実現できる。

例えば、BigQueryからCloud SQLへロードしたい場合、以下の2パターンで実装できる。

• BigQuery側のPythonモデルでCloud SQLへロード
• Cloud SQL側のPythonモデルでBigQueryからロード

ちなみに、Cloud SQLからBigQueryへロードしたい場合は、Pythonモデルを使う必要はなく、連携クエリを使用することでロードできる。

ここでは、データウェアハウス (BigQuery)とデータマート (Cloud SQL) の2つのdbtプロジェクトがあると仮定して、データマート側のPythonモデルからデータウェアハウスのデータをロードする実例を紹介します。

import os
from google.cloud import bigquery

def model(dbt, fal):
    # dbtの設定：マテリアライズをテーブルに指定
    dbt.config(materialized="table")

    client = bigquery.Client()

    query = f"""
        SELECT *
        FROM `sample_project.sample_dataset.sample_table`
    """

    # クエリの実行と結果の取得
    df = client.query(query).to_dataframe()

    return df

後は、dbt runを実行すればBigQueryからCloud SQLへデータをロードできます。

上記のコードではBigQueryのテーブルの全てのレコードを抽出していますが、増分更新も実装できます。dbt-falのPythonモデルでは増分更新をサポートしていないため、このモデルで増分レコードを抽出し、後続のモデルでそのレコードを追加する2段階構成で増分モデルを実装できます。

まとめ

dbtだけで抽出、変換、ロードを完結できるため、システムの複雑性が低減し、シンプルな構成を実現できます。しかし、2つのdbtプロジェクトを管理するなど依存関係が複雑になるデメリットもあります。