bigquery

BigQuery CLI 与 MCP 服务

CLI 与 MCP 对比

CLI 是完整接口。MCP 工具只是其中的子集。

对于下文 MCP 工具部分未列出的任何操作，默认使用 bigquery CLI。需要 Agent 使用结构化工具调用语义时，使用 MCP。

身份验证

使用 Google Cloud SDK（gcloud）ADC，无需单独进行 BigQuery 身份验证。

bigquery auth check   # verify status and scopes

默认输出格式

命令	默认格式	说明
query	JSON	包含元数据（字节数、费用、缓存命中）
datasets list	文本	易于阅读
tables list	文本	易于阅读

所有命令均接受 --format json|text|jsonl。

查询操作

# Basic (automatic dry-run + cost confirmation)
bigquery query "SELECT * FROM dataset.table LIMIT 10"

# Skip cost confirmation (required for automation)
bigquery query --yes "SELECT COUNT(*) FROM dataset.table"

# JSONL output (one object per line — for pipelines/DuckDB)
bigquery query --jsonl "SELECT * FROM dataset.table"

# Stream large results
bigquery query "SELECT * FROM large_table" --stream > results.jsonl

# From file / stdin
bigquery query --file query.sql
cat query.sql | bigquery query --stdin --yes

# Dry run (cost estimate, $0 to run)
bigquery dry-run "SELECT * FROM large_table WHERE date >= '2025-01-01'"

费用提示：每次查询前自动执行 dry-run；费用超过阈值（默认：1 GB）时会提示。响应包含 bytesProcessed、estimatedCostUSD、cacheHit。

关键标志：--yes、--format、--jsonl、--stream、--cost-threshold、--max-results、--project

重要提示：--max-results 限制的是返回的行数，而不是扫描的字节数（不会降低费用）。请改用 SQL 中的 LIMIT。

数据集操作

bigquery datasets list [--project PROJECT]
bigquery datasets describe PROJECT.DATASET
bigquery datasets create PROJECT.DATASET [--description "..." --location US --default-ttl 30]
bigquery datasets update PROJECT.DATASET --description "..."
bigquery datasets delete PROJECT.DATASET [--force --yes]

表操作

# Read
bigquery tables list PROJECT.DATASET [--max-results 20]
bigquery tables describe PROJECT.DATASET.TABLE

# Write
bigquery tables create TABLE --schema '[{"name":"id","type":"STRING"}]'
bigquery tables update TABLE --description "..."

# Insert (streaming, <1000 rows)
bigquery tables insert TABLE file.jsonl --format jsonl
cat data.jsonl | bigquery tables insert TABLE - --format jsonl

# Load (bulk, >10MB — from Cloud Storage)
bigquery tables load TABLE gs://bucket/data.csv --format csv --write-disposition WRITE_TRUNCATE

# Copy / extract / delete
bigquery tables copy SOURCE DEST [--write-disposition WRITE_TRUNCATE]
bigquery tables extract TABLE gs://bucket/output.csv --format csv
bigquery tables delete TABLE

tables load 注意事项：不接受 --yes；--write-disposition 必须使用大写（WRITE_APPEND，不能用 append）。

作业管理

bigquery jobs list [--state-filter running|done --all-users]
bigquery jobs get JOB_ID
bigquery jobs cancel JOB_ID

Python 子进程模式

# query via stdin
result = subprocess.run(
    ["bigquery", "query", "--format", "jsonl", "--yes", "--stdin"],
    input=sql, capture_output=True, text=True, check=True
)

# bulk load (no --yes for tables load)
subprocess.run(
    ["bigquery", "tables", "load", "--format", "json",
     "--write-disposition", "WRITE_APPEND", table, tmpfile],
    check=True
)

MCP 服务

启动

bigquery mcp stdio                 # read-only (default)
bigquery mcp stdio --enable-writes # with write tools
bigquery mcp http --port 3000      # HTTP mode

MCP 工具（默认只读）

在 Claude Code 中，完整标识符为 mcp__bigquery__*。

读取工具（始终可用）：

工具	说明
query	执行 SQL，自动进行 dry-run 并设置费用保护
dry_run	估算费用，不执行查询（$0）
list_datasets	列出数据集；返回 bigquery://project.dataset URI
list_tables	列出表；返回 bigquery://project.dataset.table URI
describe_table	完整架构、行数、元数据

写入工具（需要 --enable-writes）：

工具	说明
create_dataset	创建新数据集
create_table	创建表，并设置架构、分区和集群
insert_rows	流式插入（<1000 行）
load_data	从 GCS URI 加载数据（批量）
execute_ddl	运行 DDL/DML（CREATE、INSERT、UPDATE 等）

MCP 资源

表通过 bigquery://{project.dataset.table} 作为资源公开。获取资源会返回与 describe_table 相同的输出——架构、行数、元数据——且不会占用工具调用名额。

bigquery://easypost-platform.DORA.ai_attribution

工作流（MCP）

1. list_datasets → 发现数据集（返回 URI 列表）
2. list_tables → 浏览表（返回 URI 列表）
3. 读取资源 bigquery://project.dataset.table → 获取架构
4. dry_run → 估算费用
5. query → 执行

MCP 配置

{
  "mcpServers": {
    "bigquery": {
      "command": "bigquery",
      "args": ["mcp", "stdio"]
    }
  }
}

仅追加的原始表模式

• table_raw — 仅追加写入目标；绝不就地执行 UPDATE/DELETE
• table（VIEW）— 读取面；通过 ROW_NUMBER() OVER (PARTITION BY <key> ORDER BY loaded_at DESC) = 1 去重
• 始终包含 loaded_at TIMESTAMP REQUIRED；写入时设为 CURRENT_TIMESTAMP()
• 要“更新”记录：插入包含更正值的新行，并设置新的 loaded_at
• 绝不将 VIEW 作为 DML 目标——始终写入原始表

写入数据——插入与加载模式

A：tables load 会静默忽略本地文件路径

bigquery tables load 要求使用 Cloud Storage URI（gs://...）。传入本地路径会返回退出代码 0，且不打印任何内容——表不会写入，后续读取会返回 0 行。本地文件请使用 tables insert。

B：流式插入命令

bigquery tables insert --format jsonl --yes <project.dataset.table> <local-file.jsonl>

不存在 bigquery insert（不带 tables）命令——会返回“unknown command”。文件必须为 JSONL 格式。--yes 会跳过费用确认提示。

C：新表传播延迟

执行 bigquery tables create 后，元数据传播最多需要约 50 秒。立即调用 tables insert 会返回“Not found: Table”错误。请将插入操作放入重试循环中：

for attempt in range(10):
    result = subprocess.run([...tables insert...], capture_output=True, text=True)
    if result.returncode == 0:
        break
    if "Not found" in result.stderr:
        time.sleep(5)
        continue
    raise RuntimeError(result.stderr)
else:
    raise RuntimeError("table not available after 50s")

D：流式缓冲区会阻止 DELETE/TRUNCATE，持续约 90 分钟

流式插入后，DELETE FROM table WHERE TRUE 和 TRUNCATE TABLE 会失败：

UPDATE or DELETE statement over table would affect rows in the streaming buffer, which is not supported

此窗口最长持续 90 分钟。对于 WRITE_TRUNCATE 语义（每次运行时替换所有行），请删除并重新创建：

subprocess.run(["bigquery", "tables", "delete", table], capture_output=True)
_ensure_table(table)  # recreate with schema
# propagation delay applies after recreate — use retry loop from (C)

已知问题

JSONL 输出中的 DATE 列会返回 {"value": "YYYY-MM-DD"} 字典，而不是字符串。TIMESTAMP 列会返回纪元浮点数字符串（"1.760730316667E9"），而不是 ISO 字符串。

def _parse_date(v) -> str:
    if isinstance(v, dict): return str(v["value"])[:10]
    return str(v)[:10]

def _parse_ts(v) -> datetime:
    try: return datetime.fromisoformat(str(v))
    except ValueError: return datetime.fromtimestamp(float(v), tz=timezone.utc)

**数值列会以字符串形式传入。**聚合前请先转换类型：

df["col"] = pd.to_numeric(df["col"], errors="coerce").fillna(0).astype("int64")

carrier_accounts 连接导致行数膨胀：ep-core-data.raw.carrier_accounts 有约 77K 个重复 ID。连接前务必使用 SELECT DISTINCT id, ... 去重。验证方法：连接后的 COUNT() ≈ 连接前的 COUNT()。

CTE/列别名冲突：如果 CTE 与列别名同名，BQ 会将 ORDER BY col 解析为 CTE（STRUCT），从而报错“Ordering by expressions of type STRUCT is not allowed.”。请重命名列别名。

预先解析维度查询：shipments（4.2TB）未按 user_id 聚簇。连接 dim.users 以按 SF 账户筛选时，会扫描完整的日期分区（77GB+）。请改为先用低成本查询解析维度 ID，再在事实表上使用 IN 子句，无需连接。

**Salesforce created_date**是 Polytomic 同步时间戳（2025 年 2 月起），而不是 SF 创建日期。查询历史数据时，请使用 close_date 或 SF 原生日期字段。

**salesforce.tasks.created_date**是 INT64 纳秒值，需使用 TIMESTAMP_MILLIS(CAST(created_date / 1000000 AS INT64)) 转换。

UNNEST 字面量大小限制：通过 SQL 字符串中的字面量数组 UNNEST([key1, key2, ...]) 传递大量键来构建 SQL 查询时（例如 50k+ 个值），bigquery CLI 子进程会因 JSON 解析错误而失败，例如“Expecting ',' delimiter: line 1 column 1517”。SQL 本身可能语法正确，但 CLI 遇到了输入/输出限制。

解决方法：不要将大量键集嵌入 SQL 字面量中。请改为从目标表获取所有行（如果可行，可使用 REGEXP_CONTAINS 或类似条件预筛选），然后在 Python 客户端进行键匹配。extract_phab.py 中的 _build_revision_lookup 采用了相同方法：获取所有修订，在 Python 中解析工单引用，并与目标键集求交集。

快速参考

bigquery auth check
bigquery query --yes "SELECT ..."
bigquery dry-run "SELECT ..."
bigquery datasets list
bigquery tables list PROJECT.DATASET
bigquery tables describe TABLE
bigquery tables insert TABLE file.jsonl --format jsonl
bigquery tables load TABLE gs://bucket/file.csv --write-disposition WRITE_TRUNCATE
bigquery mcp stdio