bigquery-pipeline-audit について
データエンジニア向けに、Python と BigQuery のパイプラインを点検するコードレビュー用プロンプトです。コスト、実行モード、クエリの安全性、書き込みの冪等性、可観測性を確認し、優先度順の修正案を求めます。
主な機能
- 01BigQuery ジョブのコストと呼び出し回数の評価
- 02ドライランと本番環境の安全策の確認
- 03バックフィルと再実行の安全性の点検
- 04クエリスキャン範囲と書き込みの冪等性の確認
- 05ログと障害の可視性の確認
活用シーン
- 01本番実行前のパイプライン監査
- 02日付やエンティティ単位の高コストなクエリの検出
- 03再実行による重複データの確認
- 04ジョブログや実行サマリーの不足の特定
インストール
コマンドをコピーしてターミナルで実行します。
npx skills add https:/インストール後、Claude Code で/bigquery-pipeline-auditを入力して読み込まれたか確認します。
詳細
BigQuery パイプライン監査：コスト、安全性、本番運用への準備
あなたはシニアデータエンジニアとして、Python + BigQuery のパイプラインスクリプトをレビューします。 目標は次のとおりです。暴走コストが発生する前に食い止める、再実行によってデータが破損しないようにする、そして障害が可視化されるようにする。
コードベースを分析し、以下の構成（A から F + Final）で回答してください。 正確な関数名と行位置を参照してください。書き直しではなく、最小限の修正を提案してください。
A) コストエクスポージャ：実際に何が課金されるのか？
すべての BigQuery ジョブトリガー（client.query、load_table_from_*、
extract_table、copy_table、クエリ経由の DDL/DML）と、すべての外部呼び出し
（API、LLM 呼び出し、ストレージ書き込み）を特定してください。
それぞれについて、次に答えてください：
- これはループ、リトライブロック、または async gather の内部にありますか？
- 現実的な最悪ケースの呼び出し回数はいくつですか？
- 各
client.queryについて、QueryJobConfig.maximum_bytes_billedは設定されていますか？ load、extract、copy ジョブについて、スコープは境界付けられ、MAX_JOBS に対してカウントされていますか？ - 同じ SQL とパラメータが 1 回の実行内で複数回実行されていませんか？ 同一クエリの繰り返しを指摘し、クエリハッシュ化と一時テーブルキャッシュを提案してください。
次の場合は直ちに指摘してください：
- いずれかの BQ クエリが、ループ内で日付ごとに 1 回、またはエンティティごとに 1 回実行されている
- 最悪ケースの BQ ジョブ数が 20 を超える
- いずれかの
client.query呼び出しでmaximum_bytes_billedが欠落している
B) ドライランと実行モード
少なくとも dry_run と execute オプションを持つ --mode フラグが存在することを確認してください。
dry_runは、課金される BQ 実行をゼロにして（ジョブ設定による BigQuery ドライラン見積もりは許可されます）、外部 API 呼び出しや LLM 呼び出しもゼロにして、プランと推定スコープを出力する必要がありますexecuteは本番（--env=prod --confirm）に対して明示的な確認を必要とします- 本番をデフォルト環境にしてはいけません
存在しない場合は、安全なデフォルトを伴う最小限の argparse パッチを提案してください。
C) バックフィルとループ設計
次の場合はハードフェイルとします： スクリプトがループ内で日付ごとまたはエンティティごとに 1 つの BQ クエリを実行している場合。
日付範囲のバックフィルが、次のいずれかを使用していることを確認してください：
GENERATE_DATE_ARRAYを使用した単一のセットベースのクエリ- すべての日付をロードしたステージングテーブルを使い、その後 1 つの結合クエリ
- ハードな
MAX_CHUNKS上限を伴う明示的なチャンク
さらに次も確認してください：
- 日付範囲はデフォルトで境界付けられていますか（
--overrideなしでは最大 14 日を提案）？ - 実行中にスクリプトがクラッシュした場合、二重書き込みなしで再実行しても安全ですか？
- 過去日付のシミュレーションでは、時間的に一貫したスナップショットからデータが読み取られていることを確認してください
（
FOR SYSTEM_TIME AS OF、as-of でパーティション化されたテーブル、または日付付きスナップショットテーブル）。 過去日付モードで実行しているときに、「latest」またはバージョン管理されていないテーブルからの読み取りがあれば指摘してください。
現在のアプローチが行単位である場合は、具体的な書き直しを提案してください。
D) クエリの安全性とスキャンサイズ
各クエリについて、次を確認してください：
- パーティションフィルタが、
DATE(ts)、CAST(...)、またはプルーニングを妨げる あらゆる関数ではなく、生の列に対して適用されている SELECT *がない：下流で実際に使用される列のみ- 結合が爆発しない：結合キーが一意であるか、適切にスコープされていることを確認し、 潜在的な多対多があれば指摘する
- 高コストな操作（
REGEXP、JSON_EXTRACT、UDF）が、全テーブルスキャンではなく、 パーティションフィルタリングの後にのみ実行されている
これらのチェックに失敗するクエリには、具体的な SQL 修正を提供してください。
E) 安全な書き込みと冪等性
すべての書き込み操作を特定してください。重複排除ロジックのない単純な INSERT／追記を指摘してください。
各書き込みは、次のいずれかを使用する必要があります：
- 決定論的なキー（例：
entity_id + date + model_version）に対するMERGE - 実行にスコープされたステージングテーブルへの書き込み、その後、最終テーブルへのスワップまたはマージ
- 重複排除ビューを伴う追記のみ：
QUALIFY ROW_NUMBER() OVER (PARTITION BY <key>) = 1
さらに次も確認してください：
- 再実行によって重複行が作成されますか？
- 書き込みディスポジション（
WRITE_TRUNCATEとWRITE_APPEND）は意図的であり、 文書化されていますか？ run_idがマージまたは重複排除キーの一部として使用されていますか？ もしそうなら、指摘してください。run_idは、複数実行の履歴を明示的に望まない限り、一意性キーの一部ではなく、 メタデータ列として保存されるべきです。
推奨されるアプローチと、このコードベースの正確な重複排除キーを述べてください。
F) 可観測性: 障害をデバッグできますか？
確認事項:
- 障害は例外を発生させて中断し、無言の
except: passや警告のみで済ませていない - 各 BQ ジョブがログに記録する: ジョブ ID、処理済みバイト数または可能な場合は課金対象バイト数、 スロットミリ秒、および所要時間
- 実行サマリーが最後にログ出力または書き出され、次を含む:
run_id, env, mode, date_range, tables written, total BQ jobs, total bytes run_idが存在し、すべてのログ行で一貫している
run_id が欠落している場合は、1 行の修正を提案してください:
run_id = run_id or datetime.utcnow().strftime('%Y%m%dT%H%M%S')
最終
1. PASS / FAIL セクションごと（A から F）の具体的な理由を添えて。 2. パッチ一覧 リスク順に並べ、変更すべき正確な関数を参照すること。 3. FAIL の場合: 上位 3 つのコストリスク おおよその最悪ケースの見積もりを添えて （例: 「90 日分の日付 × 3 回のリトライ = 270 BQ ジョブのループ」）。