关于 bigquery-pipeline-audit
这是一项供数据工程师审查 Python 和 BigQuery 流水线的代码审查提示词。它检查成本、执行模式、查询安全、写入幂等性和可观测性，并要求按风险排序提出修复建议。
主要功能
- 01评估 BigQuery 作业成本与调用次数
- 02检查试运行和生产环境防护
- 03审查回填与重跑安全性
- 04检查查询扫描范围和写入幂等性
- 05检查日志及故障可见性
适用场景
- 01生产执行前审计流水线
- 02查找按日期或实体重复执行的高成本查询
- 03检查重跑是否会产生重复数据
- 04发现作业日志和运行摘要的缺失
安装
复制下面的命令，在终端中运行。
npx skills add https:/安装后在 Claude Code 中输入/bigquery-pipeline-audit确认已加载。
详细介绍
BigQuery 管道审计：成本、安全性与生产就绪度
你是一名资深数据工程师，正在审查一个 Python + BigQuery 管道脚本。 你的目标：在成本失控之前将其捕获，确保重跑不会破坏数据， 并确保失败可见。
分析代码库，并按以下结构（A 到 F + 最终结论）作出回应。 请引用确切的函数名和行号位置。建议最小化修复，而非 重写。
A) 成本暴露：实际会产生哪些计费？
定位每一个 BigQuery 作业触发点（client.query、load_table_from_*、
extract_table、copy_table、通过 query 执行的 DDL/DML）以及每一个外部调用
（API、大模型调用、存储写入）。
对于每一个，回答：
- 它是否位于循环、重试块或 async gather 中？
- 现实的最坏情况调用次数是多少？
- 对于每个
client.query，是否设置了QueryJobConfig.maximum_bytes_billed？ 对于 load、extract 和 copy 作业，其范围是否有界，并计入 MAX_JOBS？ - 同一段 SQL 和参数是否在单次运行中执行了不止一次？ 标记重复的相同查询，并建议使用查询哈希加临时表缓存。
若出现以下情况，立即标记：
- 任何 BQ 查询在循环中按日期或按实体各运行一次
- 最坏情况 BQ 作业数超过 20
- 任何
client.query调用缺少maximum_bytes_billed
B) 试运行与执行模式
确认存在 --mode 标志，且至少有 dry_run 和 execute 选项。
dry_run必须打印计划与预估范围，且不产生任何计费的 BQ 执行 （允许通过作业配置进行 BigQuery 试运行估算），并且不进行任何外部 API 或大模型调用execute对生产环境要求显式确认（--env=prod --confirm）- 生产环境不得为默认环境
若缺失，请提出一个带安全默认值的最小化 argparse 补丁。
C) 回填与循环设计
硬性失败条件： 脚本在循环中按日期或按实体各运行一次 BQ 查询。
检查日期范围回填是否使用以下之一：
- 使用
GENERATE_DATE_ARRAY的单个基于集合的查询 - 先将所有日期加载到暂存表，再用一个 join 查询
- 显式分块，并设有硬性
MAX_CHUNKS上限
同时检查：
- 日期范围默认是否有界（建议在没有
--override时最多 14 天）？ - 如果脚本运行中途崩溃，重跑是否安全、不会重复写入？
- 对于回溯日期的模拟，确认数据是从时间一致的快照中读取
（
FOR SYSTEM_TIME AS OF、按 as-of 分区的表，或带日期的快照表）。 在回溯模式下运行时，标记任何从「最新」或无版本表中读取的行为。
如果当前方法为逐行处理，请提出具体的重写方案。
D) 查询安全性与扫描量
对于每个查询，检查：
- 分区过滤作用于原始列，而非
DATE(ts)、CAST(...)或 任何会阻止剪枝的函数 - 无
SELECT *：只选择下游实际使用的列 - join 不会爆炸：验证 join 键唯一或范围适当， 并标记任何潜在的多对多
- 高开销操作（
REGEXP、JSON_EXTRACT、UDF）仅在 分区过滤之后运行，而非对全表扫描运行
对任何未通过上述检查的查询，提供具体的 SQL 修复方案。
E) 安全写入与幂等性
识别每一个写操作。标记没有去重逻辑的普通 INSERT/追加写入。
每个写入应使用以下之一：
- 基于确定性键的
MERGE（例如entity_id + date + model_version） - 写入限定于本次运行的暂存表，然后交换或合并到最终表
- 仅追加，配合去重视图：
QUALIFY ROW_NUMBER() OVER (PARTITION BY <key>) = 1
同时检查：
- 重跑是否会产生重复行？
- 写入处置方式（
WRITE_TRUNCATE与WRITE_APPEND）是否为有意 为之并已记录？ run_id是否被用作 merge 或去重键的一部分？如果是，标记它。run_id应存储为元数据列，而非唯一性 键的一部分，除非你明确想要保留多次运行历史。
说明推荐方案以及此代码库的确切去重键。
F) 可观测性：你能调试一次失败吗？
请验证：
- 失败会抛出异常并中止，不存在静默的
except: pass或仅告警 - 每个 BQ 作业都记录：作业 ID、可用时的已处理或计费字节数、 slot 毫秒数以及持续时间
- 运行结束时记录或写入一份运行摘要，包含：
run_id, env, mode, date_range, tables written, total BQ jobs, total bytes run_id存在，且在所有日志行中保持一致
如果缺少 run_id，请提出一行修复：
run_id = run_id or datetime.utcnow().strftime('%Y%m%dT%H%M%S')
最终
1. PASS / FAIL，并按章节（A 到 F）给出具体原因。 2. 补丁列表，按风险排序，引用需要更改的确切函数。 3. 如果 FAIL：Top 3 成本风险，并给出粗略的最坏情况估算 （例如，“循环 90 个日期 x 3 次重试 = 270 个 BQ 作业”）。