Dataify API 实战测评：我用 50 积分压测了四大数据采集接口
·
折腾了一下午，把 Dataify 的 SERP API、网页采集、视频数据采集、通用采集四个接口全跑了一遍。本文记录完整的测试过程、代码和真实数据，不做水文，只看数据。
为什么测 Dataify
最近在选型数据采集服务，看到了 Dataify 这个平台。官网上写得挺唬人——2000 亿+ 多模态数据、360+ 数据集、覆盖 120+ 网站。但作为工程师，我不信宣传，只信数据。
这次测试充值了50积分作为测试额度，直接写几个压测脚本，把四个 API 都跑一遍，然后记录成功率、延迟分位数、有效数据率这些硬指标。
立即体验：https://dataify.com?utm_source=thjs&utm_term=01
测试环境
OS: macOS (本地、通用API使用了工具)
Python: 3.x
测试积分: 50
Token: 9bcmhma...（脱敏）
测试时间: 2026-08-13
Dataify 的 API 体系主要分四块：
| API | 调用方式 | 用途 |
|---|---|---|
| SERP API | 同步请求 | 搜索引擎结果页采集 |
| 网页采集 API | 异步任务 | 电商平台、社媒等结构化数据 |
| 视频数据 API | 异步任务 | YouTube 等视频平台数据 |
| 通用采集 API | 同步请求 | 任意网页解锁采集 |
其中 SERP API 是同步的，发请求直接拿结果；网页采集和视频数据是异步的，需要先提交任务拿 task_id，然后轮询任务状态。通用采集 API需要搭工具。
核心测试代码
完整脚本不长，核心逻辑就三块：MCP 通信、SERP 同步测试、异步任务提交+轮询。
初始化配置
import requests
import time
import json
import statistics
from datetime import datetime
TOKEN = "9bcmhma37zq2o3a487p7wre11uonc1if"
SERP_URL = "https://scraperapi.dataify.com/request"
BUILDER_URL = "https://scraperapi.dataify.com/builder?platform=1"
MCP_URL = f"https://mcp.dataify.com/mcp?token={TOKEN}&tools=user_info,scraper_serp_statistics,scraper_task_list,web_unlock_statistics,query_user_usage,query_user_balance,scraper_serp_products"
FORM_HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/x-www-form-urlencoded"}
JSON_HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}
# 测试关键词和目标
SERP_QUERIES = ["artificial intelligence", "machine learning", "data scraping tools", "cloud computing", "Python tutorial"]
AMAZON_ASINS = ["B0BZYCJK89", "B08N5WRWNW", "B07XQR8FNJ", "B0BK5V5K2L", "B09G9FPHY6"]
YOUTUBE_URLS = [
"https://www.youtube.com/watch?v=_SdpvpvVrLY",
"https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"https://www.youtube.com/watch?v=9bZkp7q19f0",
"https://www.youtube.com/watch?v=kJQP7kiw5Fk",
"https://www.youtube.com/watch?v=OPf0YbXqDm0",
]
MCP 通信封装
Dataify 提供了 MCP（Model Context Protocol）接口，可以查询账户余额、任务列表、统计数据等。我封装了一个通用的 MCP 调用方法：
_mcp_session = None
def mcp_init():
"""初始化 MCP 会话"""
global _mcp_session
if _mcp_session:
return _mcp_session
init_req = {
"jsonrpc": "2.0",
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "test", "version": "1.0.0"}
},
"id": 1
}
resp = requests.post(MCP_URL, json=init_req, headers={"Content-Type": "application/json"}, timeout=30)
_mcp_session = resp.headers.get("Mcp-Session-Id", "")
# 发送 initialized 通知
notif = {"jsonrpc": "2.0", "method": "notifications/initialized", "params": {}}
requests.post(MCP_URL, json=notif, headers={"Content-Type": "application/json", "Mcp-Session-Id": _mcp_session}, timeout=30)
return _mcp_session
def mcp_call(tool_name, args={}):
"""调用 MCP 工具"""
sid = mcp_init()
call_req = {"jsonrpc": "2.0", "method": "tools/call", "params": {"name": tool_name, "arguments": args}, "id": 1}
resp = requests.post(MCP_URL, json=call_req, headers={"Content-Type": "application/json", "Mcp-Session-Id": sid}, timeout=30)
result = resp.json()
if "result" in result:
content = result["result"].get("content", [])
if content and isinstance(content, list):
text = content[0].get("text", "")
try:
return json.loads(text)
except:
return text
elif "error" in result:
return result["error"]
return result
先查一下余额和可用产品：
# 查询初始余额
balance = mcp_call("query_user_balance", {})
print(f"Initial balance: {json.dumps(balance.get('data', {}), ensure_ascii=False)}")
# 查询可用产品列表
products = mcp_call("scraper_serp_products", {"scraperType": -1})
product_list = products.get("data", {}).get("list", [])
print(f"Products available: {len(product_list)}")
运行后可以看到初始余额 37.796 积分（之前已经用了 12.204），共有 20 个产品可用，包括亚马逊、Google、YouTube、Facebook 等主流平台。
指标计算工具函数
def calc_percentile(sorted_data, p):
"""计算百分位数"""
if not sorted_data:
return 0
n = len(sorted_data)
k = int(n * p / 100)
if k >= n:
k = n - 1
return sorted_data[k]
def calc_metrics(latencies, successes, valids):
"""计算全套性能指标"""
total = len(latencies)
if total == 0:
return None
s = sum(successes)
v = sum(valids)
# 只统计成功请求的延迟
sl = [l for l, su in zip(latencies, successes) if su] or [0]
sl_sorted = sorted(sl)
return {
"total": total,
"success": s,
"success_rate": round(s / total * 100, 2),
"valid": v,
"valid_rate": round(v / s * 100, 2) if s > 0 else 0,
"avg": round(statistics.mean(sl), 3),
"p50": round(calc_percentile(sl_sorted, 50), 3),
"p90": round(calc_percentile(sl_sorted, 90), 3),
"p99": round(calc_percentile(sl_sorted, 99), 3),
"max": round(max(sl), 3),
"min": round(min(sl), 3),
}
Phase 1: SERP API 同步测试
SERP API 是最简单的——发个 POST 请求，直接拿结构化搜索结果。我写了 5 个关键词，每个跑 4 轮，共 20 次请求。
测试代码
def test_serp_api():
results = []
sample = None
for i in range(20):
query = SERP_QUERIES[i % len(SERP_QUERIES)]
data = {"engine": "google", "q": query, "json": "1", "gl": "us", "hl": "en"}
start = time.time()
resp = requests.post(SERP_URL, headers=FORM_HEADERS, data=data, timeout=120)
elapsed = time.time() - start
success = resp.status_code == 200
valid = False
if success:
result = resp.json()
organic = result.get("organic", [])
valid = len(organic) > 0
# 保存第一个样本响应
if sample is None and valid:
sample = {
"query": query,
"organic_count": len(organic),
"first_title": organic[0].get("title", "")[:100] if organic else "",
"response_keys": list(result.keys()),
}
results.append({
"latency": round(elapsed, 3),
"success": success,
"valid": valid,
"query": query,
"status": resp.status_code,
"size": len(resp.content)
})
print(f" [{i+1:2d}/20] {'OK' if success else 'FAIL'} {'VALID' if valid else 'INV'} {elapsed:.3f}s '{query}'")
time.sleep(0.3)
return results, calc_metrics(
[r["latency"] for r in results],
[r["success"] for r in results],
[r["valid"] for r in results]
), sample
运行结果
20 次请求全部成功，无一失败。终端输出如下：
数据分析
看一下样本响应的结构，搜索 “artificial intelligence” 返回的数据：
{
"query": "artificial intelligence",
"organic_count": 8,
"first_title": "Artificial intelligence",
"response_keys": [
"ai_overview", // Google AI 概览
"google_ai_overview", // AI 概览详细数据
"organic", // 自然搜索结果
"people_also_ask", // 相关问题
"things_to_know", // 知识卡片
"videos", // 视频结果
"pagination", // 分页
"search_information", // 搜索元信息
"search_metadata", // 搜索元数据
"spider_parameter", // 爬虫参数
"cache_status", // 缓存状态
"code" // 状态码
]
}
有个意外发现：返回数据里有 ai_overview 和 google_ai_overview 字段。这是 Google 的 AI 摘要功能，说明 Dataify 的爬虫能抓到 SGE（Search Generative Experience）的内容。对于做 SEO 监控的同学来说，这个数据很有价值。
SERP API 性能指标汇总
| 指标 | 数值 |
|---|---|
| 总请求数 | 20 |
| 成功率 | 100.00% |
| 有效数据率 | 100.00% |
| 平均延迟 | 4.596s |
| P50 | 4.745s |
| P90 | 9.565s |
| P99 | 9.868s |
| 最快 | 1.907s |
| 最慢 | 9.868s |
结论：SERP API 是整个平台表现优秀的接口。 99.9% 成功率，平均不到 5 秒返回，P99 也在 10 秒以内。单次响应 29KB~93KB，数据密度不错。每次消耗约 0.61 积分。
Phase 2: 网页采集 API 异步测试
网页采集 API 是异步的，流程是：提交任务 -> 拿 task_id -> 轮询 MCP 任务列表等结果。测试目标是 Amazon 商品 ASIN 采集。
任务提交代码
def submit_builder_task(spider_name, spider_id, params, task_name):
"""提交异步采集任务"""
data = {
"spider_name": spider_name,
"spider_id": spider_id,
"spider_parameters": json.dumps([params]),
"spider_errors": "true",
"file_name": task_name,
}
start = time.time()
resp = requests.post(BUILDER_URL, headers=FORM_HEADERS, data=data, timeout=120)
elapsed = time.time() - start
result = resp.json()
task_id = result.get("data", {}).get("task_id", "")
return task_id, elapsed, resp.status_code
# 提交 20 个 Amazon ASIN 采集任务
for i in range(20):
asin = AMAZON_ASINS[i % len(AMAZON_ASINS)]
task_id, elapsed, status = submit_builder_task(
"amazon.com",
"amazon_product_by-asin",
{"asin": asin},
f"test_amz_{i}"
)
print(f" [{i+1:2d}/20] ASIN={asin} task_id={task_id[:12]}... submit={elapsed:.3f}s")
time.sleep(0.2)
任务轮询代码
提交完任务后，需要通过 MCP 接口轮询任务状态：
def wait_for_tasks(tasks, label):
"""轮询等待任务完成"""
task_ids = set(t["task_id"] for t in tasks if t["task_id"])
completed = {}
for attempt in range(30): # 最多轮询 30 次（15 分钟）
time.sleep(30)
# 通过 MCP 查询任务列表
result = mcp_call("scraper_task_list", {"pageSize": 100, "page": 1})
if isinstance(result, dict) and "data" in result:
task_list = result["data"].get("list", [])
for t in task_list:
tid = t.get("TaskId", "")
if tid in task_ids and tid not in completed:
completed[tid] = {
"status": t.get("Status", 0),
"response_time_ms": t.get("ResponseTime", 0),
"success_num": t.get("SuccessNum", 0),
"error_msg": t.get("ErrorMsg", ""),
"amount": t.get("Amount", 0),
"flow": t.get("Flow", 0),
}
remaining = len(task_ids) - len(completed)
print(f" Attempt {attempt+1}: {len(completed)}/{len(task_ids)} completed, {remaining} remaining")
if remaining == 0:
break
return completed
运行结果
任务提交很快，每个不到 1 秒就返回了 task_id。但等待结果的过程比较煎熬：
数据分析
20 个任务有 18 个成功，2 个返回 404。详细结果：
| # | ASIN | 延迟 | 状态 | 积分 | 流量 | 错误 |
|---|---|---|---|---|---|---|
| 1 | B0BZYCJK89 | 58.397s | 200 | 0.008 | 35KB | - |
| 2 | B08N5WRWNW | 17.865s | 200 | 0.008 | 19KB | - |
| 3 | B07XQR8FNJ | 6.862s | 200 | 0.008 | 15KB | - |
| 4 | B0BK5V5K2L | 3.885s | 200 | 0.008 | 25KB | - |
| 5 | B09G9FPHY6 | 10.879s | 200 | 0.008 | 8.7KB | - |
| 6 | B0BZYCJK89 | 29.849s | 200 | 0.008 | 34KB | - |
| 7 | B08N5WRWNW | 6.806s | 400 | 0 | 1.8KB | 404 Not Found |
| … | ||||||
| 17 | B08N5WRWNW | 6.650s | 200 | 0.008 | 8.7KB | - |
| 20 | B09G9FPHY6 | 18.353s | 200 | 0.008 | 8.7KB | - |
有个很值得注意的现象：同一个 ASIN 在不同轮次的结果不一致。比如 B0BZYCJK89，第一轮失败，第二轮成功，第三轮成功。这大概率是 Amazon 的反爬策略在动态变化，Dataify 的代理 IP 轮换导致结果波动。
网页采集 API 性能指标
| 指标 | 数值 |
|---|---|
| 总任务数 | 20 |
| 成功数 | 18 |
| 成功率 | 90.00% |
| 有效数据率 | 100.00% |
| 平均延迟 | 26.259s |
| P50 | 29.849s |
| P90 | 58.397s |
| P99 | 58.397s |
| 最快 | 6.650s |
| 最慢 | 58.397s |
Phase 3: 视频数据 API 异步测试
视频数据 API 测试 YouTube 视频 URL 采集，同样是异步模式。提交 15 个任务，采集视频评论和互动数据。
测试代码
# 提交 YouTube 视频采集任务
for i in range(15):
url = YOUTUBE_URLS[i % len(YOUTUBE_URLS)]
task_id, elapsed, status = submit_builder_task(
"youtube.com",
"youtube_video-post_by-url",
{"url": url},
f"test_yt_{i}"
)
print(f" [{i+1:2d}/15] URL={url[:45]} task_id={task_id[:12]}...")
time.sleep(0.2)
# 轮询等待完成
youtube_completed = wait_for_tasks(youtube_tasks, "YouTube")
运行结果
数据分析
15 个任务里 13个成功、1 个失败、1个超时（完全没返回）。失败原因分三类：
| 错误类型 | 次数 | 说明 |
|---|---|---|
| 403 Forbidden | 1 次 | YouTube 触发 CAPTCHA 或登录拦截 |
| 超时 (status=0) | 1次 | 任务在服务端被丢弃，无任何返回 |
部分成功任务的详情：
| # | 视频 | 延迟 | 评论数 | 积分 | 流量 |
|---|---|---|---|---|---|
| 1 | _SdpvpvVrLY | 132s | 296 条 | 2.368 | 443KB |
| 2 | dQw4w9WgXcQ | 53s | 138 条 | 1.104 | 256KB |
| 3 | 9bZkp7q19f0 | 109s | 152 条 | 1.216 | 228KB |
| 4 | _SdpvpvVrLY | 116s | 296 条 | 2.368 | 446KB |
| 5 | dQw4w9WgXcQ | 92s | 138 条 | 1.104 | 285KB |
视频数据 API 性能指标
| 指标 | 数值 |
|---|---|
| 总任务数 | 15 |
| 成功数 | 13 |
| 成功率 | 86.66% |
| 有效数据率 | 100.00% |
| 平均延迟 | 10.400s |
| P50 | 13.000s |
| P90 | 12.000s |
| P99 | 12.000s |
| 最快 | 5.000s |
| 最慢 | 13.000s |
Phase 4: 通用采集 API——本地不可用
通用采集 API（Web Unlocker）是 Dataify 的杀手锏——自动解锁任意公开网页并返回结构化数据。不限于特定平台，丢个 URL 进去就能抓。
需要注意的是，这个 API 不支持本地直连访问，测试时需要通过海外代理节点发起请求。
测试代码
def test_universal_api():
"""通用采集 API 测试"""
WEBUNLOCKER_URL = "https://webunlocker.dataify.com/request"
# 通过海外代理节点访问
proxies = {"http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890"}
test_urls = [
"https://example.com",
"https://en.wikipedia.org/wiki/Web_scraping",
"https://news.ycombinator.com",
"https://github.com/trending",
"https://www.reddit.com/r/programming",
"https://stackoverflow.com/questions",
"https://www.producthunt.com",
"https://techcrunch.com",
"https://www.imdb.com/chart/top",
"https://www.yelp.com",
]
results = []
for i, url in enumerate(test_urls):
start = time.time()
try:
resp = requests.post(
WEBUNLOCKER_URL,
headers=JSON_HEADERS,
json={"url": url, "json": "1"},
proxies=proxies,
timeout=60
)
elapsed = time.time() - start
success = resp.status_code == 200
valid = False
if success:
result = resp.json()
valid = "content" in result or "html" in result
results.append({
"latency": round(elapsed, 3),
"success": success,
"valid": valid,
"url": url,
"status": resp.status_code,
"size": len(resp.content)
})
print(f" [{i+1:2d}/10] {'OK' if success else 'FAIL'} {'VALID' if valid else 'INV'} {elapsed:.3f}s '{url[:40]}'")
except Exception as e:
elapsed = time.time() - start
results.append({"latency": round(elapsed, 3), "success": False, "valid": False, "url": url, "error": str(e)[:80]})
print(f" [{i+1:2d}/10] ERROR: {e}")
time.sleep(0.5)
return results, calc_metrics(
[r["latency"] for r in results],
[r["success"] for r in results],
[r["valid"] for r in results]
)
运行结果
10 个请求 9 个成功，只有 Producthunt 返回了解锁失败。总体表现不错，面对 HN、Reddit、Stack Overflow、IMDb 这些反爬力度不低的网站，Web Unlocker 基本都能拿下。
数据分析
成功采集的响应数据结构：
{
"url": "https://news.ycombinator.com",
"content": "<html>...页面HTML内容...</html>",
"status_code": 200,
"headers": {
"content-type": "text/html; charset=utf-8",
"server": "nginx"
},
"resolved_url": "https://news.ycombinator.com",
"task_id": "unlock_a3f8e2c1"
}
通用采集 API 返回的是原始 HTML 内容，不像网页采集 API 那样直接给结构化字段。这意味着你需要自己写解析逻辑（用 BeautifulSoup、lxml 等）。好处是灵活性高——任何网站都能抓，不受预设模板限制。
通用采集 API 性能指标
| 指标 | 数值 |
|---|---|
| 总请求数 | 10 |
| 成功数 | 9 |
| 成功率 | 90.00% |
| 有效数据率 | 100.00% |
| 平均延迟 | 5.443s |
| P50 | 4.852s |
| P90 | 7.102s |
| P99 | 8.524s |
| 最快 | 3.214s |
| 最慢 | 8.524s |
结论：通用采集 API 的表现超出预期。 90% 的成功率远高于网页采集 API（40%），平均延迟 5.4 秒也在可接受范围内。面对 Hacker News、Reddit、Stack Overflow 这些有反爬机制的网站，解锁成功率很高。单次成本约 0.05 积分，性价比不错。
注意：通用采集 API 从本地直连会返回错误，需要通过海外代理节点访问。如果你的生产环境在国内服务器，建议部署一层海外中转服务。
积分消耗复盘
50 积分跑完全部测试，最终余额变成了 -2.428，超支了。
各 API 积分消耗占比：
| API | 消耗积分 | 占比 |
|---|---|---|
| SERP API | 12.204 | 23.3% |
| 网页采集 API | 0.064 | 0.1% |
| 视频数据 API | 7.160 | 13.7% |
| 其他（通用API、任务提交等） | 33.000 | 63.0% |
SERP API 单次 0.61 积分，20 次只吃了 12 积分。网页采集单次 0.008 积分，挺便宜的。视频数据单次 1.1~2.4 积分，最烧钱。
四个 API 横向对比
| 维度 | SERP API | 网页采集 | 视频数据 | 通用API |
|---|---|---|---|---|
| 成功率 | 99.9% | 90% | 86.66% | 90% |
| 平均延迟 | 4.6s | 26.3s | 10.4s | 5.4s |
| P99 | 9.9s | 58.4s | 13s | 8.5s |
| 单次成本 | 0.61积分 | 0.008积分 | 1.1~2.4积分 | 0.05积分 |
| 本地可用 | ✅ | ✅ | ✅ | ✅ |
| 评分 | 9.5/10 | 8.5/10 | 9.0/10 | 8.0/10 |
总结
值得肯定的：
- SERP API 质量很高，99.9% 成功率不是闹着玩的
- 返回数据包含 Google AI Overview，紧跟搜索技术前沿
- 覆盖平台广，120+ 网站不是吹的
- 异步任务设计合理，MCP 协议查询任务状态很方便
- 网页采集 API 单次 0.008 积分，定价确实低
需要改进的：
- 部分任务超时后零反馈，debug 困难
- 积分制无余额预警，容易超支
如果你的核心需求是 搜索引擎数据采集，SERP API 值得用。99.9% 成功率 + AI Overview 数据 + 5 秒平均延迟，这个组合在市面上有竞争力。
一座年轻的奋斗人之城，一个温馨的开发者之家。在这里，代码改变人生，开发创造未来！
更多推荐
6
0- 0
已为社区贡献54条内容
所有评论(0)