用Playwright打造可靠的企业级采集方案--从单机验证到集群化落地

原创

jackcode

发布于 2025-09-23 10:24:42

11510

代码可运行

文章被收录于专栏：爬虫资料爬虫资料

运行总次数：0

代码可运行

爬虫代理

为什么要做这个项目（背景与动机）

在公司做数据产品时，我们常常遇到三个痛点：脚本跑不稳、页面渲染抓不到数据、以及规模化后调度和重试逻辑变得难以维护。最开始通常是一两个 Python 脚本在开发机上跑通，但到了生产环境，问题就会接踵而来：单机负载、被目标站点限速、以及不同页面结构带来的脆弱性。

本项目的目标是把一个可验证的单机 Playwright 爬虫，逐步演进成能在多台机器上并行工作的集群化方案，并把关键实现、思路与示例代码记录下来，方便求职/展示或直接在企业内复用。

我们要拿到什么数据（数据目标）

本文以招聘网站为示例，目标字段很常见也很实用，便于后续做分析：

职位名称（title）
公司名（company）
城市 / 工作地点（city）
薪资（salary）
发布时间（publish_time）

这些字段既能做趋势分析，也能用于构建岗位搜索引擎或行业报告。

技术栈与选型理由（为什么选 Playwright）

Playwright：相比传统的 requests，它能原生处理 JS 渲染、等待节点、拦截资源等，稳定性和可控性更好。
代理 IP（示例：亿牛云）：高频访问需要代理来分摊流量并降低被限制的概率。
Redis（任务队列） + 多进程/多机器：用于实现任务分发、去重与伸缩。
MongoDB / Elasticsearch：分别用于原始数据落盘和后续的检索/分析/可视化。

这种组合能在保证可维护性的同时兼顾扩展与性能。

核心实现（从单机到分布式的演进）

1）单机版：先把逻辑跑通

这是最直接的版本，用来确认页面结构、JS 渲染及提取逻辑是否稳健。代码示例（异步 Playwright）：

# single_playwright.py
import asyncio
from playwright.async_api import async_playwright

# ====== 代理配置（示例：亿牛云） ======
proxy_host = "proxy.16yun.cn"
proxy_port = "3100"
proxy_user = "16YUN"
proxy_pass = "16IP"

async def crawl_page(keyword: str, page_num: int = 1):
    """
    单页抓取示例函数：访问并解析职位卡片，返回字典列表
    """
    async with async_playwright() as pw:
        # 启动浏览器（无头/有头可根据调试需要切换）
        browser = await pw.chromium.launch(headless=True,
                                           proxy={
                                               "server": f"http://{proxy_host}:{proxy_port}",
                                               "username": proxy_user,
                                               "password": proxy_pass
                                           })
        page = await browser.new_page()
        url = f"https://example-job-site.com/search?kw={keyword}&p={page_num}"
        await page.goto(url, timeout=30000)
        # 等待职位列表出现（根据实际页面调整选择器）
        await page.wait_for_selector('.job-card', timeout=15000)

        cards = await page.query_selector_all('.job-card')
        results = []
        for c in cards:
            # 下面的选择器只是示例，实际按目标站点调整
            title = await c.query_selector_eval('.job-title', 'el => el.textContent.trim()')
            company = await c.query_selector_eval('.company', 'el => el.textContent.trim()')
            salary = await c.query_selector_eval('.salary', 'el => el.textContent.trim()')
            city = await c.query_selector_eval('.city', 'el => el.textContent.trim()')
            results.append({
                "title": title,
                "company": company,
                "salary": salary,
                "city": city
            })

        await browser.close()
        return results

if __name__ == "__main__":
    rows = asyncio.run(crawl_page("Python 爬虫", page_num=1))
    for r in rows:
        print(r)

说明：先把单页、单关键词跑通，再处理翻页、异常与重试逻辑。单机阶段以稳定为先，别急着并发。

2）改造为可并发的单机版本（充分利用单机资源）

当单页逻辑稳定后，我们会把单页函数包装成任务池，利用 asyncio 并发多个标签页（或多个浏览器实例）来提高吞吐。示例要点：

限制并发并设置合理的超时
处理网络异常与元素不存在的情况（捕获并重试）
记录日志与关键指标（成功率、耗时）

（此处略去并发包装的完整代码，实际项目中可用 asyncio.Semaphore 控制并发数。）

3）走向分布式：Redis 作任务队列（生产者 / 消费者模型）

把任务放到 Redis 列表或流（stream）里，多台机器从队列里取任务并执行。示例消费端伪代码：

# worker.py (伪代码，需加异常处理与日志)
import asyncio
import aioredis
from single_playwright import crawl_page  # 引入前面实现的单页函数

REDIS_URL = "redis://127.0.0.1:6379/0"
TASK_QUEUE = "job_tasks"

async def worker_loop():
    redis = await aioredis.from_url(REDIS_URL)
    while True:
        task = await redis.lpop(TASK_QUEUE)
        if not task:
            await asyncio.sleep(1)
            continue
        # 假设任务是 JSON 字符串，包含 keyword 和 page_num
        task = task.decode("utf-8")
        # 这里简单演示，生产环境请解析 JSON 并做失败重试、幂等处理
        keyword = task
        print("取到任务：", keyword)
        results = await crawl_page(keyword)
        # 结果入库／发送到下游（例如 MongoDB/Elasticsearch）
        # save_results(results)

关键点：

任务应包含去重标识（比如 URL 哈希或唯一 ID）以避免重复抓取
设计任务超时与重试策略（比如 N 次重试后写入失败队列由人工检查）
监控队列深度、消费速率以判断是否需要扩容

运营与工程化要点（实用经验）

代理策略：不要把所有流量都打到一个代理 IP，建议轮换或使用多个出口；对静态资源设置拦截，减少不必要带宽消耗。
资源拦截：Playwright 支持 route.abort() 拦截图片、视频、广告等资源，显著减轻网络压力。
健康检查：为每个爬虫进程加上心跳上报（Prometheus / Pushgateway）便于发现挂死或慢速任务。
错误收集：记录页面快照（在非敏感场景），异常堆栈与请求/响应，便于定位问题。
容器化：把爬虫打包为 Docker 镜像，用 Kubernetes 编排能显著简化部署与弹性伸缩。
数据质量控制：入库前做字段校验、去重与时间窗口过滤，避免脏数据污染分析维度。

数据落盘与可视化（展示成就）

落库示例（MongoDB 文档结构）：

{
  "title": "Python 爬虫工程师",
  "company": "某某科技",
  "salary": "20k-35k",
  "city": "上海",
  "publish_time": "2025-09-01",
  "source": "example-job-site",
  "crawl_at": "2025-09-23T10:00:00"
}

落库后可以做的展示（列举思路）：