用 SerpBase 采集的数据落成 Parquet,再用 DuckDB 做分析

用 SERP API 采回来的数据,绝大多数人就是往磁盘上一扔,一堆 keyword_2026-10-10_p1.json。真要用的时候才发现:想知道"这个月哪些关键词的 top10 换过 30% 以上",得写一堆循环读 JSON、逐条比对,跑一次十分钟。

我后来把采集结果统一落成 Parquet、用 DuckDB 查,同样的问题几秒钟出结果。这篇讲这套轻量数据栈怎么搭,代码可以直接抄。

为什么是 Parquet + DuckDB

先说选型理由,不讲空话:

• Parquet 是列式存储:只读你要的列,数据量大了优势明显;自带压缩,磁盘占用通常比原始 JSON 小好几倍;schema 明确,不会出现某天字段突然变了却没人发现。
• DuckDB 是嵌入式分析数据库:单文件、零部署,直接 SELECT * FROM 'xxx.parquet' 就能查;支持标准 SQL,团队里谁都会用;不需要起服务、不需要装驱动。
• 两者都不依赖外部服务:对个人开发者和中小团队,不用维护数仓,本地就能跑。

第一步:采集时就把字段拍平

关键在采集阶段就把嵌套结构拍平成"一行一个结果"的宽表,别等到分析时再拆。

import json, time, requests, pyarrow as pa, pyarrow.parquet as pq
from pathlib import Path

BASE = "https://api.serpbase.dev"
HEADERS = {"X-API-Key": "YOUR_API_KEY", "Content-Type": "application/json"}
OUT = Path("serp_lake"); OUT.mkdir(exist_ok=True)

def collect_one(keyword, gl="us", hl="en", page=1, device=None):
body = {"q": keyword, "gl": gl, "hl": hl, "page": page}
if device:
body["device"] = device
r = requests.post(f"{BASE}/google/search", headers=HEADERS, json=body, timeout=20)
payload = r.json()
if payload.get("status") != 0:
return []
rows = []
for item in payload.get("organic") or []:
rows.append({
"keyword": keyword,
"gl": gl,
"hl": hl,
"page": page,
"device": device or "default",
"rank": item.get("rank"),
"title": item.get("title"),
"link": item.get("link"),
"snippet": item.get("snippet"),
"date": item.get("date"),
# 信封字段也带上,后面算成本直接聚合
"request_id": payload.get("request_id"),
"elapsed_ms": payload.get("elapsed_ms"),
"credits_charged": payload.get("credits_charged"),
"collected_at": int(time.time()),
})
return rows

四个设计点:

第二步:批量落成 Parquet,按天分区

import datetime

def collect_batch(keywords, gl="us", hl="en"):
all_rows = []
for kw in keywords:
all_rows.extend(collect_one(kw, gl=gl, hl=hl))
time.sleep(0.25) # 匀速配速,别打满配额窗口
if not all_rows:
return None
# 按天分区:一天一个文件,避免单文件过大,也方便增量追加
day = datetime.date.today().isoformat()
path = OUT / f"day={day}" / f"serp-{int(time.time())}.parquet"
path.parent.mkdir(parents=True, exist_ok=True)
table = pa.Table.from_pylist(all_rows)
pq.write_table(table, path, compression="zstd")
print(f"wrote {len(all_rows)} rows -> {path}")
return path

按天分区(day=YYYY-MM-DD/)是刻意的:后续做"按天对比"时,DuckDB 可以直接用通配符只扫相关分区,不用全表扫描。

第三步:用 DuckDB 做三件最常问的事

装完 DuckDB(pip install duckdb)就能直接查 Parquet 文件,不用先建表。

问题一:本月哪些关键词的 top10 换过 30% 以上?

import duckdb

con = duckdb.connect()

sql = """
WITH ranked AS (
SELECT keyword,
date_trunc('day', to_timestamp(collected_at)) AS day,
list(link ORDER BY rank) AS top10
FROM 'serp_lake/day=*/serp-*.parquet'
WHERE rank <= 10
GROUP BY keyword, day
),
pairs AS (
SELECT keyword,
day,
top10,
lag(top10) OVER (PARTITION BY keyword ORDER BY day) AS prev
FROM ranked
)
SELECT keyword,
day,
round(100.0 * len(list_intersect(top10, prev)) / 10, 1) AS overlap_pct
FROM pairs
WHERE prev IS NOT NULL
AND len(list_intersect(top10, prev)) < 7 — 重合不足 7 条 = 换过 30% 以上
ORDER BY overlap_pct ASC
"""
print(con.execute(sql).fetchdf())

list_intersect 是 DuckDB 的列表函数,直接算两个 top10 集合的交集,不用自己写循环。这一条 SQL 替你省掉了一整套 diff 脚本。

问题二:成本与延迟按端点和地区拆开看。

sql2 = """
SELECT gl,
count(DISTINCT keyword) AS keywords,
sum(credits_charged) AS total_credits,
round(avg(elapsed_ms), 0) AS avg_ms,
quantile_cont(elapsed_ms, 0.95) AS p95_ms
FROM 'serp_lake/day=*/serp-*.parquet'
GROUP BY gl
ORDER BY total_credits DESC
"""

这里体现出信封字段落地的价值:成本、地域、延迟三个维度一条 SQL 全出来,不用回头翻日志。

问题三:一个新词最早出现在哪天?

sql3 = """
SELECT keyword,
min(to_timestamp(collected_at)) AS first_seen
FROM 'serp_lake/day=*/serp-*.parquet'
GROUP BY keyword
ORDER BY first_seen DESC
LIMIT 20
"""

成本账

/google/search 是 1 credit 一次。上面的采集函数一个关键词一页平均返回 10 行左右,也就是说 100 个关键词落盘大约 1000 行,消耗约 100 credits。落 Parquet 和 DuckDB 查询都在本地,零额外成本。

真正的省钱点在复用:数据一旦落成 Parquet,后面无论换多少种分析口径、改多少次阈值,都是本地重算,不用重新调接口。这也是我把"采集"和"分析"彻底分开的原因——采集只跑一次,分析随便跑。

字段口径和端点费率我是按 SerpBase 的搜索端点文档 核对的;信封里的 credits_charged 和 elapsed_ms 建议照着上面的方式落进表,后面做对账和性能分析都省事。

下一步

挑三个你常查的关键词,按上面的代码跑三天,看看 Parquet 文件有多大、DuckDB 查一次多快。你会发现在这个量级上,一套轻量数据栈完全够用,不需要上任何真正意义上的数据仓库。