dbt-sf-to-bq-translator

dbt Snowflake から BigQuery へのトランスレーター

担当する内容：

1. 方言の変換：Snowflake の dbt SQL モデルを、Google Cloud 上の標準化された BigQuery SQL に変換します。
2. 標準化とコンプライアンス：各ファイルの先頭に著作権ヘッダーを付けること、明示的な型キャスト、標準化された JSON 抽出、_extracted_at を使用した QUALIFY による重複排除など、Google 固有の標準を適用します。
3. ワークフローへの統合：dbt の Jinja 構文を保持し、最終的な BigQuery 互換モデルを保存します。

migration_plan/[mig_prefix]/tasks.md に記載された指示に従ってください。作業中に進捗を追記し、最後にタスクファイルへ概要を記載して、人間の監督者が進捗を確認できるようにします。エラーから復旧する際は、タスクファイルを確認してください。

帰属情報

必ずお読みください：gcloud、REST API、または MCP コマンドを実行する前に、必ず使用状況の帰属情報をお読みください。ターミナルで直接実行するコマンド（生成スクリプト、Terraform、ユーザーコード内では実行しない場合）では、gcloud の前に CLOUDSDK_METRICS_REQUEST_ATTRIBUTION="agent-skills/1.0 (skill:dbt-sf-to-bq-translator)" を付け、User-Agent ヘッダーに agent-skills/1.0 (skill:dbt-sf-to-bq-translator) を追加してください。

前提条件と環境のセットアップ

変換を開始する前に、Google Cloud 環境が適切に設定されていることを確認してください。

1. Google Cloud SDK：まだインストールされていない場合は、Google Cloud SDK をインストールします。
2. 認証：CLI セッションを認証します。

   gcloud auth login
   gcloud auth application-default login
   

3. プロジェクトの設定：有効な Google Cloud プロジェクトを設定します。

   gcloud config set project {project_id}
   

4. 請求先アカウント：対象プロジェクトに有効な Google Cloud 請求先アカウントが関連付けられていることを確認します。
5. 必要な API の有効化：BigQuery、Migration、Storage の各サービスが有効になっていることを確認します。

   gcloud services enable bigquerymigration.googleapis.com storage.googleapis.com bigquery.googleapis.com
   

6. リージョンの選択：希望するコンピューティング／BigQuery リージョンを設定します（推奨されるデフォルト値：us-central1 または us）。Google Cloud のロケーションを参照してください。

   gcloud config set compute/region us-central1
   

手順

• 初期化とセットアップ：mig_prefix が定義されていない場合、またはユーザーが新しい変換プロジェクトの開始を希望する場合は、まずユーザーに以下を必ず確認してください。

  1. 移行プロジェクト名（例：my_migration_project）。
  2. Snowflake SQL ファイルを含む入力ディレクトリ。
  3. BigQuery SQL ファイルの保存先となる出力ディレクトリ。
  4. 変換アセットのステージングに使用する Cloud Storage バケット名。
  5. Google Cloud リージョン（例：us または eu）。
  6. （任意）ソースデータベースのメタデータ（columns.csv や tables.csv など）を含むディレクトリまたは .zip ファイルのローカルパス。 情報を受け取ったら、移行手順を表す未完了タスクを含むチェックリストファイルを migration_plan/[mig_prefix]/tasks.md に作成します。

• 同梱スクリプトによる自動実行： 同梱の変換スクリプト scripts/bulk_translate_via_gcloud.py を使用して、決定的なエンドツーエンドの移行を実行します。

  python3 scripts/bulk_translate_via_gcloud.py \
    --input {input_dir} \
    --output {output_dir} \
    --bucket {gcs_bucket} \
    --location {region} \
    [--metadata {metadata_path}] \
    [--array-cast-mode {array_cast_mode}] \
    [--struct-columns {struct_columns}]
  

  フラグ	デフォルト	説明
  --input	(必須)	ソースの Snowflake dbt .sql および .yml ファイルを含むローカルディレクトリです。
  --output	(必須)	変換後の BigQuery dbt .sql および .yml ファイルの出力先ディレクトリです。
  --bucket	(必須)	変換の入力と出力を一時保存するために使用する Cloud Storage バケット名です。
  --location	us	BigQuery Translation Service の実行に使用する Google Cloud リージョンです（例：us、eu）。
  --metadata	None	ソースの tables.csv および columns.csv を含むディレクトリまたは .zip アーカイブの省略可能なパスです。
  --array-cast-mode	array_string	BQMS の ARRAY<JSON> キャストに対する変換先です。デフォルトでは array_string（ARRAY<STRING>）を使用します。ソーステーブルのスキーマに応じて、代替手段として string（STRING）、json_extract_string_array、または preserve を使用できます。
  --struct-columns	None	JSON_VALUE からネイティブのドット記法（col.field）に変換する、追加の BigQuery STRUCT 列名をカンマ区切りで指定する省略可能なリストです（例：struct_col,record_col）。

  scripts/ に含まれる移行ツールは、以下の一連の処理を実行します。

  • scripts/bulk_translate_via_gcloud.py：前処理、Cloud Storage へのアップロード、BigQuery Translation Service の呼び出し、ダウンロード、後処理、YAML 設定のコピーを自動化し、エンドツーエンドの一括移行を統括します。
  • scripts/dbt_translator.py：config(...) 用の決定論的な AST パーサー、Jinja プレースホルダーのマスキングと復元、JSON／STRUCT／ARRAY 抽出のサニタイズ、未実装オペコードのクリーンアップ、マクロの監査、大文字・小文字と結合の標準化、著作権ヘッダーの適用を含む、コア変換ライブラリです。

• 詳細な翻訳ライフサイクル（スクリプトによって実行）：

  1. プレースホルダーを使用して標準 SQL にコンパイル（翻訳前）：
     • 元の dbt .sql ファイルを読み込みます。各ファイルの先頭にある {{ config(...) }} ヘッダーブロックを抽出して削除します。
     • BigQuery Translation Service で構文エラーが発生しないよう、dbt マクロ呼び出しを標準 SQL に準拠したプレースホルダー識別子に置き換えます。
       • {{ source('src_name', 'table_name') }} を _DBT_SOURCE_src_name_DBTSEP_table_name_ に置き換えます。
       • {{ ref('model_name') }} を _DBT_REF_model_name_ に置き換えます。
     • Snowflake の角括弧記法（col['prop'] -> col:prop）、::variant:field、Jinja の {% else %}／ループ内の UNION ALL 分岐を正規化し、翻訳前に Jinja の波括弧（{{ ... }}）を取り除くことで、トランスパイラーが 100％有効な Snowflake 方言の SQL を処理できるようにします。
  2. SQL ファイルを分離する：
     • 前処理済みの Jinja を含まないファイルを、Cloud Storage へのアップロードに備えてステージング入力ディレクトリに保存します。
  3. メタデータを前処理し、BigQuery Translation Service で SQL を翻訳する：
     • メタデータのパスが指定されている場合、検出された dbt モデル／ソースに一致するテーブルエントリを columns.csv と tables.csv 内のプレースホルダー名に対応付け、名前空間の解決エラーを防ぐためカタログ名をクリアし、metadata.zip にまとめて Cloud Storage にアップロードします。
     • ステージング SQL ファイルを Cloud Storage にアップロードします。 gcloud storage cp <staging_input_dir>/*.sql gs://[YOUR_BUCKET]/migration_input/
     • ソースを snowflakeDialect、ターゲットを bigqueryDialect として指定する migration_config.yaml を作成します（メタデータが指定されている場合は、schemaPath が metadata.zip を指すようにします）。
     • 翻訳ワークフローを開始します。 gcloud bq migration-workflows create --location=<region> --config-file=migration_config.yaml --no-async
     • 翻訳済みの GoogleSQL ファイルを Cloud Storage からダウンロードします。 gcloud storage cp gs://[YOUR_BUCKET]/migration_output/*.sql <translated_output_dir>/
  4. プレースホルダーを復元し、dbt のロジックを再度組み込む：
     • 翻訳済みの BigQuery SQL ファイルに対して、高度な後処理を行います。
       • AST ベースの設定変換：元の {{ config(...) }} ブロックを解析し、copy_grants、transient、secure などの Snowflake 固有のパラメーターを削除します。pre_hook と post_hook をサニタイズし、ALTER ICEBERG TABLE ... REFRESH、UNSET SECURE、COPY INTO などの無効な Snowflake コマンドを取り除き、有効なものは保持します。
       • 参照リゾルバー（名前空間の解決）：SQL 内の FROM 句と JOIN 句にあるハードコードされた Snowflake データベース／スキーマのテーブルパスをスキャンし、検出されたプロジェクトのモデルおよびソースを照合して、ネイティブの dbt {{ ref(...) }} または {{ source(...) }} マクロに戻します。
       • マクロと構文の監査：すべての {{ ... }} Jinja 式をスキャンし、カスタムまたは許可リストにないデータベース固有のマクロについて警告を記録します。また、これらのブロックを Snowflake 固有の構文（例：::date、dateadd、to_date）について監査し、見落とされたりマスクされたりした可能性のあるものを示す警告コメントをファイル内に直接記載します。
       • 括弧の対応を考慮した SQL のエッジケースのサニタイズ：括弧の対応を解析して、日付キャストのサフィックス（::date -> CAST(... AS DATE)）、日時キャストのサフィックス（::timestamp -> CAST(... AS TIMESTAMP)）、浮動小数点数キャストのサフィックス（::float -> CAST(... AS FLOAT64)）、ネストされた dateadd(...) 呼び出し、および制御ブロックの内外にある区間（- interval '5 month'）を変換します。CTE の周囲にずれた Jinja タグを修正し、16MB の substr(string(...), 1, 16777216) ラッパーを削除します。
       • 配列型と STRUCT アクセスの正規化：型指定のない BQMS の ARRAY<JSON> キャストを ARRAY<STRING>（または設定済みの --array-cast-mode のターゲット）に変換し、STRUCT 列に対する json_value を直接のドット記法（col.field）に、ARRAY<STRUCT> 列に対するものを col[SAFE_OFFSET(idx)].field に変換し、error_unimplemented オペコードをクリーンアップします。
       • 著作権ヘッダーの配置：必須の Google 著作権ヘッダーを、設定ブロックより前のファイルの最上部に追加します。
  5. 新しい BigQuery dbt ファイルに書き込み、YAML 設定をコピーする：
     • ディレクトリ構造を維持しながら、新しく組み立てた BigQuery 互換ファイルを保存します。また、入力ディレクトリ内のすべての .yml／.yaml ファイルを出力ディレクトリにコピーします。

• 最後に、タスクファイルに概要を記載します。

• 引き渡す前に、成果物についてユーザーの承認を求めます。ユーザーから必要な変更を受けたら、適用します。

• タスクファイルでタスクを完了済みにします。

必須事項と行動規則

1. 必須のファイルヘッダー

翻訳するファイルはすべて、先頭に以下の正確なヘッダーを必ず記載してください。

# Copyright 2026 Google. This software is provided as-is, without warranty or
# representation for any use or purpose. Your use of it is subject to your
# agreement with Google.

2. 標準化された JSON と配列の抽出

BigQuery SQL では Snowflake のコロン記法（:）を使用しないでください。BigQuery Standard SQL は半構造化フィールドへのアクセスにコロン構文を認識しないため、構文エラーが発生します。半構造化データの検索には BigQuery の標準関数を使用し、詳しいパターンとエラーへの対処方法については dbt_migration_patterns.md を参照してください。

• スカラー値と文字列化された JSON の抽出：ネイティブの JSON 列には JSON_VALUE(json_col, '$.path') を使用します（配列インデックスの前にドットを入れず、例：$.items[1]）。文字列化された JSON 列には CAST(JSON_EXTRACT_SCALAR(SAFE.PARSE_JSON(CAST(source.col AS STRING)), '$.path') AS TYPE) を使用します。厳密な型指定が必要な場合は、型指定した抽出を SAFE_CAST(... AS INT64|BOOL|TIMESTAMP|NUMERIC) で囲みます。
• 配列の抽出、展開、型指定なし配列（::array）：JSON_EXTRACT_STRING_ARRAY または JSON_EXTRACT_ARRAY を使用し、CROSS JOIN UNNEST(...) で展開します。Snowflake の ::array から BQMS の ARRAY<JSON> キャストへの変換では、デフォルトで ARRAY<STRING> にマッピングします（ソーススキーマに応じて STRING または JSON_EXTRACT_STRING_ARRAY を使用してください。詳しくは dbt_migration_patterns.md のセクション 3 を参照）。
• JSON タイムスタンプ（'Z' で終わる ISO-8601 形式）：JSON のタイムスタンプ抽出には TIMESTAMP をキャストします（CAST(JSON_EXTRACT_SCALAR(json_col, '$.updated_at') AS TIMESTAMP)）。'Z' で終わる ISO-8601 形式の文字列は CAST(... AS DATETIME) で実行時エラーになるためです。下流で DATETIME が必要な場合は、いったん TIMESTAMP にキャストしてから変換します（詳しくは dbt_migration_patterns.md のセクション 5 を参照）。
• BigQuery の STRUCT 列と ARRAY<STRUCT> 列：ネイティブの STRUCT 列には直接ドット記法（col.field、例：source.struct_col.count）を使用します。ARRAY<STRUCT> 列には source.col[SAFE_OFFSET(0)].field（または JSON_VALUE(TO_JSON(source.col), '$[0].field')）を使用します。JSON_VALUE() と CAST(... AS JSON) は STRUCT 型を受け付けないためです（詳しくは dbt_migration_patterns.md のセクション 6 を参照）。
• Snowflake のブラケット記法（col['field']）と error_unimplemented：前処理中に col['field'] を col:field に正規化し、残っている BQMS の error_unimplemented または 16MB の substr(string(...), 1, 16777216) ラッパーは、STRUCT 列の場合は col.field に、JSON 列の場合は JSON_VALUE(col, '$.field') に置き換えます（詳しくは dbt_migration_patterns.md のセクション 7〜8 を参照）。

3. 明示的な型安全性

• 比較：結合またはフィルターの両辺の型が一致しない場合は、必ず両辺で SAFE_CAST を使用します。必要に応じて、比較の安全性を広く確保できる AS STRING を使用します。
• ID フィールド：すべてのシステム ID（例：ticket_id、user_id）には INT64 を優先して使用します。
• NULL の処理：
  • プレースホルダー列には、必ず明示的な型キャストを使用します：SAFE_CAST(NULL AS TYPE)。

4. 文字列関数と日付関数の指定

• 切り詰め：LEFT(col, length) または SUBSTR(col, 1, length) を使用します。
• 検索：REGEXP_CONTAINS の代わりに LOWER(col) LIKE '%pattern%' を使用します。
• 日付の加算：DATE_ADD(SAFE_CAST(col AS DATETIME), INTERVAL num HOUR) を使用します。

5. 重複排除と結合の必須ルール

• 重複排除：ソースモデルで主キーによる重複排除が必要な場合：
  • ルール：ベース CTE で row_number() over (partition by [PRIMARY_KEY] order by [TIMESTAMP] desc) as rn 列を使用し、その CTE に直接 qualify rn = 1 を適用します。
• 結合：カスタムフィールドまたは属性テーブル（例：exploded_array パターン）との結合には LEFT JOIN を使用し、レコードが失われないようにします。

6. Jinja 構文を保持する

{{ config(...) }}、{{ ref(...) }}、{{ source(...) }} は変更しないでください。{% if is_incremental() %} ブロックが機能する状態を保ってください。ソースファイルに記載されていない限り、過去のデータを結合する処理や、その他のカスタムマクロ／テーブルを追加しないでください。

出力

migration_plan/[mig_prefix]/translated_models/ に、BigQuery 互換の dbt SQL モデルを出力してください。翻訳したファイルは、構造パターンと方言の書式に厳密に従う必要があります。 翻訳例については、dbt_migration_patterns.md を参照してください。

制約

ルートエージェントに引き渡す前に、まず翻訳したモデルについてユーザーの承認を得てください。ユーザーの要望を反映した後、ルートエージェントに引き渡してください。