Google Cloud：用 BigQuery 公开数据集查一个 Python 包最近 7 天按国家/地区的下载量，导出 CSV

用 BigQuery 公开数据集查一个 Python 包最近 7 天按国家/地区的下载量，导出 CSV。脚本先 dry run 估算扫描量，超过你设的上限就停下；真正执行时再用 maximum_bytes_billed 兜底，避免一条 SQL 扫出意外账单。

先看清成品

脚本用假客户端跑了 2 个测试并通过（超限时只做 dry run、正常时传参数和计费上限）；在本机用真实 SDK（google-cloud-bigquery 3.46.2）运行，因为没有凭证在认证阶段报 DefaultCredentialsError，说明调用路径正确。没有用真实项目执行查询，扫描量和结果以你的运行为准。

注意：Google Cloud 按用量计费，价格以官方价格页和账单为准，本文不列金额。Google Cloud 不在中国大陆提供服务，请在服务可用的地区按条款注册使用，不要用代理或借用他人账号绕过地区限制，也不要共享账号。

做完之后你会得到什么？

需要准备哪些环境？

文件结构

21-google-cloud-B-src/
├── pypi_stats.py
└── test_pypi_stats.py

第 1 步：了解数据集

bigquery-public-data.pypi.file_downloads 每行是一次下载，表按 timestamp 分区，数据量很大。查询必须带时间范围，扫描量才会只覆盖相关分区。字段说明见 PyPI 官方文档。

第 2 步：脚本

pypi_stats.py：

"""查询某个 PyPI 包最近 N 天按国家/地区的下载量，先 dry run 估算扫描量，超过上限就不执行。"""
import argparse, csv, sys
from google.cloud import bigquery

SQL = """
SELECT country_code, COUNT(*) AS downloads
FROM `bigquery-public-data.pypi.file_downloads`
WHERE file.project = @project
  AND DATE(timestamp) BETWEEN DATE_SUB(CURRENT_DATE(), INTERVAL @days DAY) AND CURRENT_DATE()
GROUP BY country_code
ORDER BY downloads DESC
LIMIT 20
"""
GB = 1024 ** 3

def run(client, project, days, max_gb, out):
    params = [bigquery.ScalarQueryParameter("project", "STRING", project),
              bigquery.ScalarQueryParameter("days", "INT64", days)]
    dry = client.query(SQL, job_config=bigquery.QueryJobConfig(
        dry_run=True, use_query_cache=False, query_parameters=params))
    gb = dry.total_bytes_processed / GB
    print(f"预计扫描 {gb:.2f} GiB（上限 {max_gb} GiB）")
    if gb > max_gb:
        print("超过上限，已停止。缩短 --days 或调高 --max-gb。"); return 2
    job = client.query(SQL, job_config=bigquery.QueryJobConfig(
        query_parameters=params, maximum_bytes_billed=int(max_gb * GB)))
    rows = list(job.result())
    with open(out, "w", newline="", encoding="utf-8-sig") as f:
        w = csv.writer(f); w.writerow(["country_code", "downloads"])
        for r in rows: w.writerow([r["country_code"], r["downloads"]])
    print(f"写出 {out}：{len(rows)} 行，实际计费 {(job.total_bytes_billed or 0) / GB:.2f} GiB")
    return 0

if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("package"); ap.add_argument("--days", type=int, default=7)
    ap.add_argument("--max-gb", type=float, default=50); ap.add_argument("--out", default="downloads.csv")
    ap.add_argument("--billing-project", help="运行查询并计费的项目 ID")
    a = ap.parse_args()
    if not 1 <= a.days <= 30: sys.exit("--days 应为 1-30")
    sys.exit(run(bigquery.Client(project=a.billing_project), a.package.lower(), a.days, a.max_gb, a.out))

要点：

• @project、@days 是查询参数，包名不会被拼进 SQL。
• dry run 不收费，返回 total_bytes_processed。
• maximum_bytes_billed 是 BigQuery 端的硬上限，超过时查询直接失败而不是收费。

第 3 步：测试

test_pypi_stats.py：

from types import SimpleNamespace as NS
import csv, pypi_stats as m

class Fake:
    def __init__(self, gb): self.gb, self.calls = gb, []
    def query(self, sql, job_config):
        self.calls.append(job_config)
        if job_config.dry_run: return NS(total_bytes_processed=int(self.gb * m.GB))
        return NS(result=lambda: [{"country_code": "US", "downloads": 10}, {"country_code": "CN", "downloads": 3}],
                  total_bytes_billed=int(self.gb * m.GB))

def test_guard_stops():
    f = Fake(80); assert m.run(f, "requests", 7, 50, "x.csv") == 2 and len(f.calls) == 1

def test_runs(tmp_path):
    f = Fake(12); out = tmp_path / "d.csv"
    assert m.run(f, "requests", 7, 50, out) == 0
    assert f.calls[1].maximum_bytes_billed == 50 * m.GB
    p = {q.name: q.value for q in f.calls[1].query_parameters}; assert p == {"project": "requests", "days": 7}
    assert list(csv.reader(open(out, encoding="utf-8-sig")))[1] == ["US", "10"]

pip install google-cloud-bigquery pytest
python -m pytest -q

检查结果：2 passed。

第 4 步：真实运行（本文未执行）

gcloud auth application-default login
python pypi_stats.py requests --days 7 --max-gb 50 --billing-project 你的项目ID

预期输出格式：

预计扫描 xx.xx GiB（上限 50.0 GiB）
写出 downloads.csv：20 行，实际计费 xx.xx GiB

先用 --days 1 看估算值，再决定天数。

调试示例

本机没有凭证时实际报错：

google.auth.exceptions.DefaultCredentialsError: Your default credentials were not found.
To set up Application Default Credentials, see https://cloud.google.com/docs/authentication/external/set-up-adc ...

运行 gcloud auth application-default login 后重试。

出错了怎么办？

模板规则

• 所有查询先 dry run。
• 每个查询都设 maximum_bytes_billed。
• 参数化查询，不拼接用户输入。

项目资源

BigQuery 按扫描量计费，写错一个条件就可能扫全表。把 dry run 和计费上限写进脚本，查询就有了刹车，可以放心交给别人用。