如何提升基于BeautifulSoup与Scraper API的Pandas补数爬虫效率？

一、现有操作的问题点 #

你的代码逻辑上能实现需求，但存在几个细节问题和效率瓶颈：

1. DataFrame定位错误：df.at[row,'column_name']写法有误，df.iterrows()返回的是(索引, 行数据)元组，应该用index而非row来定位，正确写法是df.at[index, 'column_name']。
2. 无异常处理机制：未处理请求失败、页面目标元素缺失的情况，一旦某条请求出错或元素找不到，整个循环会直接中断。
3. 串行执行效率极低：iterrows()本身是DataFrame中效率较低的遍历方式，加上同步请求串行等待，1600条URL会产生大量无效等待时间。

二、效率优化方案 #

针对1600条URL的爬取场景，可从以下维度优化：

1. 改用Scraper API批量请求 #

Scraper API支持批量提交最多100个URL，大幅减少HTTP请求的握手开销，比单条请求快数倍。示例代码：

import requests
import pandas as pd
from bs4 import BeautifulSoup

base_url = 'https://www.example.com/'
api_key = 'a_key'

# 生成所有目标URL
urls = [base_url + str(date) for date in df['date']]
# 按每100个URL分组
url_batches = [urls[i:i+100] for i in range(0, len(urls), 100)]

results = []
for batch in url_batches:
    payload = {
        'api_key': api_key,
        'urls': ','.join(batch),
        'keep_headers': 'true'
    }
    response = requests.post('http://api.scraperapi.com/batch', data=payload)
    batch_results = response.json()
    
    for res in batch_results:
        if res['status_code'] == 200:
            soup = BeautifulSoup(res['response'], 'html.parser')
            try:
                value = soup.find("div", {"class": "a class"}).get_text(strip=True)[-2:]
                results.append(value)
            except AttributeError:
                results.append(None)  # 元素不存在时填充空值
        else:
            results.append(None)

# 一次性赋值给DataFrame，避免逐行写入开销
df['column_name'] = results

2. 异步并发请求 #

如果批量接口不满足需求，用aiohttp实现异步请求，同时发起多个请求（注意不要超过Scraper API的并发限制），示例：

import aiohttp
import asyncio
import pandas as pd
from bs4 import BeautifulSoup

base_url = 'https://www.example.com/'
api_url = 'http://api.scraperapi.com'
api_key = 'a_key'

async def fetch(session, target_url):
    params = {'api_key': api_key, 'url': target_url}
    try:
        async with session.get(api_url, params=params, timeout=10) as response:
            if response.status == 200:
                html = await response.text()
                soup = BeautifulSoup(html, 'html.parser')
                return soup.find("div", {"class": "a class"}).get_text(strip=True)[-2:]
            else:
                return None
    except Exception as e:
        print(f"请求失败 {target_url}: {e}")
        return None

async def main():
    urls = [base_url + str(date) for date in df['date']]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    df['column_name'] = results

asyncio.run(main())

3. 优化DataFrame遍历逻辑 #

避免使用iterrows()，直接通过df['date']生成URL列表，减少逐行操作的性能损耗。

4. 缓存重复请求 #

如果df['date']存在重复值，先对日期去重，爬取一次后通过映射填充到DataFrame：

# 提取唯一日期
unique_dates = df['date'].unique()
date_value_map = {}

for date in unique_dates:
    target_url = base_url + str(date)
    # 发起请求并解析（省略请求代码）
    date_value_map[date] = parsed_value

# 映射回原DataFrame
df['column_name'] = df['date'].map(date_value_map)

5. 合理配置请求参数 #

• 给请求添加超时时间，避免长时间等待无响应的请求；
• 确认Scraper API套餐的并发限制，不要超过阈值导致请求被限流；
• 若目标页面是动态渲染的，添加render=true参数（需对应套餐支持），确保获取完整渲染内容。

6. 添加重试机制 #

对失败的请求自动重试，避免因临时网络问题浪费爬取进度，可使用tenacity库实现：

from tenacity import retry, stop_after_attempt, wait_exponential
import requests

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_url(target_url):
    payload = {'api_key': 'a_key', 'url': target_url}
    response = requests.get('http://api.scraperapi.com', params=payload, timeout=10)
    response.raise_for_status()
    return response.content

内容的提问来源于stack exchange，提问作者pottele