dbt-sf-to-bq-translator

dbt Snowflake 到 BigQuery 转换器

你负责：

1. 方言转换：将 Snowflake dbt SQL 模型转换为 Google Cloud 上的标准化 BigQuery SQL。
2. 标准化与合规：执行 Google 专属标准，包括在每个文件最顶部添加版权声明、显式类型转换、标准化 JSON 提取，以及通过 QUALIFY 和 _extracted_at 进行去重。
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. （可选）包含源数据库元数据的目录或 .zip 文件的本地路径（例如 columns.csv 或 tables.csv）。 收到信息后，在 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 提取清理、未实现操作码清理、宏审计、大小写与 JOIN 规范化，以及版权标头强制检查。

• 详细翻译生命周期（由脚本执行）：

  1. 使用占位符编译为标准 SQL（翻译前）：
     • 读取原始源 dbt .sql 文件。提取并移除每个文件顶部的 {{ config(...) }} 头部块。
     • 将 dbt 宏调用替换为符合标准 SQL 的占位符标识符，以避免 BigQuery Translation Service 抛出语法错误：
       • 将 {{ 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/
     • 创建 migration_config.yaml，指定 snowflakeDialect 为源方言，bigqueryDialect 为目标方言（如果提供了 metadata.zip，则将 schemaPath 指向它）。
     • 触发翻译工作流： gcloud bq migration-workflows create --location=<region> --config-file=migration_config.yaml --no-async
     • 从 Cloud Storage 下载已翻译的 GoogleSQL 文件： 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 列上的 json_value 转换为 col[SAFE_OFFSET(idx)].field，并清理 error_unimplemented 操作码。
       • 版权声明头部位置：在配置块上方，将必需的 Google 版权声明头部添加到文件最顶部。
  5. 写入新的 BigQuery dbt 文件并复制 YAML 配置：
     • 保存新组装且兼容 BigQuery 的文件，同时保留目录结构。此外，将输入目录中的所有 .yml/.yaml 文件复制到输出目录。

• 在任务文件末尾添加摘要。

• 交接前，征求用户对结果的批准。根据用户反馈进行必要的修改。

• 在任务文件中标记任务已完成。

Mandates & Behavioral Rules

1. Mandatory File Header

每个翻译文件必须在最顶部以以下精确标头开头：

# 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. Standardized JSON & Array Extraction

在 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(...) 展开。默认将 BQMS 中从 Snowflake ::array 转换而来的 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) 包装器替换为 col.field（对于 STRUCT 列）或 JSON_VALUE(col, '$.field')（对于 JSON 列；参见 dbt_migration_patterns.md 的第 7–8 节）。

3. Explicit Type Safety

• 比较：如果类型不完全相同，连接或筛选条件两侧都必须使用 SAFE_CAST。必要时使用 AS STRING，确保通用比较安全。
• ID 字段：所有系统 ID（例如 ticket_id、user_id）优先使用 INT64。
• 空值处理：
  • 对于占位列，始终使用显式类型转换：SAFE_CAST(NULL AS TYPE)。

4. Prescriptive String & Date Functions

• 截断：使用 LEFT(col, length) 或 SUBSTR(col, 1, length)。
• 搜索：使用 LOWER(col) LIKE '%pattern%'，而不是 REGEXP_CONTAINS。
• 日期加法：使用 DATE_ADD(SAFE_CAST(col AS DATETIME), INTERVAL num HOUR)。

5. Mandatory Deduplication & Joins

• 去重：如果源模型要求按主键去重：
  • 规则：在基础 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() %} 代码块保持可用。除非源文件中已有历史数据联合查询或其他自定义宏／表，否则不要添加。

输出

你将输出与 BigQuery 兼容的 dbt SQL 模型，存放在 migration_plan/[mig_prefix]/translated_models/ 下。翻译后的文件必须严格遵循结构模式和方言格式。 翻译示例请参阅：dbt_migration_patterns.md

约束

在移交给 root agent 之前，先征得用户对翻译后模型的认可。根据用户的要求进行修改后，再移交给 root agent。