Links

GitHub issueにも上がっている

公式Docにも明示されている。
https://docs.getdbt.com/docs/build/incremental-models

根本的な問題: dbtの問題というよりは、BQがnested column追加のDDLに対応していない (ref)

つまり、dbtでの解決は難しく、bq commandやbq APIを使用する必要がある

解決法

手動

手動でやるなら bq コマンドでschema変更を行う

現在のschema取得

bq show \
--schema \
--format=prettyjson \
mydataset.mytable > /tmp/myschema.json

schema temp file /tmp/myschema.jsonに新しいカラムの追加

schemaの更新

bq update mydataset.mytable /tmp/myschema.json

スクリプト

import json
import argparse
from google.auth import impersonated_credentials, default
from google.cloud import bigquery


def diff_schemas(old_schema, new_schema, prefix=""):
    diffs = []

    old_fields = {field["name"]: field for field in old_schema}
    new_fields = {field["name"]: field for field in new_schema}

    for name, old_field in old_fields.items():
        full_field = f"{prefix}{name}"
        if name not in new_fields:
            diffs.append(f"Field removed: {full_field}")
        else:
            new_field = new_fields[name]

            if old_field.get("type") != new_field.get("type"):
                diffs.append(f"Type changed for field {full_field}: {old_field.get('type')} -> {new_field.get('type')}")

            if old_field.get("mode") != new_field.get("mode"):
                diffs.append(f"Mode changed for field {full_field}: {old_field.get('mode')} -> {new_field.get('mode')}")

            if old_field.get("type") == "RECORD":
                old_nested = old_field.get("fields", [])
                new_nested = new_field.get("fields", [])
                diffs.extend(diff_schemas(old_nested, new_nested, prefix=full_field + "."))

    for name, new_field in new_fields.items():
        full_field = f"{prefix}{name}"
        if name not in old_fields:
            diffs.append(f"Field added: {full_field}")

    return diffs


def get_bigquery_client(project, impersonate_service_account=None):
    """Get a BigQuery client, optionally using impersonated credentials.

    Args:
        project (str): GCP project ID.
        impersonate_service_account (str, optional): Service account email to impersonate.

    Returns:
        bigquery.Client: A BigQuery client.
    """
    if impersonate_service_account:
        # Use the default credentials to create impersonated credentials.
        source_credentials, _ = default()
        target_credentials = impersonated_credentials.Credentials(
            source_credentials=source_credentials,
            target_principal=impersonate_service_account,
            target_scopes=["https://www.googleapis.com/auth/cloud-platform"],
            lifetime=3600,
        )
        return bigquery.Client(project=project, credentials=target_credentials)
    else:
        return bigquery.Client(project=project)


def load_current_schema(
    project: str, dataset_id: str, table_id: str, schema_json: str, impersonate_service_account: str = ""
):
    """Load the current schema of a BigQuery table and save it to a JSON file.

    Args:
        project (str): GCP project ID.
        dataset_id (str): BigQuery dataset ID.
        table_id (str): BigQuery table ID.
        schema_json (str): Path to the JSON file where the schema will be saved.
    """
    # Initialize BigQuery client.
    client = get_bigquery_client(project, impersonate_service_account)

    # Get the table.
    table_ref = client.dataset(dataset_id).table(table_id)
    table = client.get_table(table_ref)

    # Convert current schema (list of SchemaField objects) into a list of dictionaries.
    current_schema = [field.to_api_repr() for field in table.schema]
    with open(schema_json, "w") as f:
        json.dump(current_schema, f, indent=2)


def update_table_schema(
    project: str,
    dataset_id: str,
    table_id: str,
    schema_json: str,
    dryrun: bool = True,
    impersonate_service_account: str = "",
):
    """Update table schema with the specified JSON file.

    Args:
        project (str): GCP project ID.
        dataset_id (str): BigQuery dataset ID.
        table_id (str): BigQuery table ID.
        schema_json (str): Path to the JSON file with the schema definition.
        dryrun (bool): Dry run (do not update the schema).

    Returns:
        None
    """

    # Initialize BigQuery client.
    client = get_bigquery_client(project, impersonate_service_account)

    # Get the table.
    table_ref = client.dataset(dataset_id).table(table_id)
    table = client.get_table(table_ref)

    # Convert current schema (list of SchemaField objects) into a list of dictionaries.
    current_schema = [field.to_api_repr() for field in table.schema]

    # Load the complete schema from the JSON file.
    with open(schema_json, "r") as f:
        new_schema_dict = json.load(f)

    # Convert new schema dictionaries to SchemaField objects.
    new_schema = [bigquery.SchemaField.from_api_repr(field) for field in new_schema_dict]

    added_fields = diff_schemas(current_schema, new_schema_dict)

    if not added_fields:
        print(f"Schema is up-to-date for table {project}.{dataset_id}.{table_id}")
        return

    if dryrun:
        # Identify new fields that are not in the current schema.
        print(f"Fields to be added: {added_fields}")
    else:
        # Assign the new schema to the table and update it.
        table.schema = new_schema
        client.update_table(table, ["schema"])
        print(f"Successfully updated schema for table {dataset_id}.{table_id}")
        return


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Update a BigQuery table schema by adding nested fields using a patch JSON file."
    )
    parser.add_argument("--impersonate_service_account", help="Service account email to impersonate")
    parser.add_argument("--project", required=True, help="GCP project ID")
    parser.add_argument("--dataset", required=True, help="BigQuery dataset ID")
    parser.add_argument("--table", required=True, help="BigQuery table ID")
    parser.add_argument(
        "--action", required=True, choices=["update", "load"], default="load", help="Action to perform (default: load)"
    )
    parser.add_argument("--schema_json", required=True, help="Path to the JSON file with the schema definition")
    parser.add_argument(
        "--no-dryrun", action="store_false", dest="dryrun", help="Perform actual update (default is dry run)"
    )

    args = parser.parse_args()
    if args.action == "load":
        load_current_schema(args.project, args.dataset, args.table, args.schema_json, args.impersonate_service_account)
    elif args.action == "update":
        update_table_schema(
            args.project, args.dataset, args.table, args.schema_json, args.dryrun, args.impersonate_service_account
        )
    else:
        raise ValueError("Invalid action. Use 'update' or 'load'.")

usage: manage_schema.py [-h] [--impersonate_service_account IMPERSONATE_SERVICE_ACCOUNT] --project PROJECT --dataset DATASET --table TABLE --action
                        {update,load} --schema_json SCHEMA_JSON [--no-dryrun]

Update a BigQuery table schema by adding nested fields using a patch JSON file.

options:
  -h, --help            show this help message and exit
  --impersonate_service_account IMPERSONATE_SERVICE_ACCOUNT
                        Service account email to impersonate
  --project PROJECT     GCP project ID
  --dataset DATASET     BigQuery dataset ID
  --table TABLE         BigQuery table ID
  --action {update,load}
                        Action to perform (default: load)
  --schema_json SCHEMA_JSON
                        Path to the JSON file with the schema definition
  --no-dryrun           Perform actual update (default is dry run)

現在のスキーマファイルを取得

python manage_schema.py --project <project> --dataset <dataset> --table <table> --action load --schema_json tmp/schema.json

スキーマファイルの更新

変更を確認 (dryrun (default))

python manage_schema.py --project <project> --dataset <dataset> --table <table> --action update --schema_json tmp/schema.json

Schema変更を適用

python manage_schema.py --project <project> --dataset <dataset> --table <table> --action update --schema_json tmp/schema.json --no-dryrun

dbt-python

どうしてもdbt内で実行したい場合は、dbt-pythonでモデルを書くことで、incremental table且つスキーマ変更があれば更新するというモデルを作成することが可能

実行方法

手動

bqコマンドや上のScriptを使った手動のオペレーション

Airflow

dbtをAirflowなどで実行している場合AirflowのDAGに入れることができる

GitHub Actions

実際の運用を考えるとGitHub Actionsで実行できるようにしておけると楽