如何在Python中高效导出与处理大型backlink数据集？

一、解决内存占用问题：流式处理+磁盘级去重 #

现有方案把所有反向链接加载到内存，同时用内存集合维护去重数据，数据量突破50k后必然出现内存溢出。改用流式处理（边获取数据边写入文件），同时用磁盘存储的索引替代内存集合实现去重，这里用轻量的SQLite做示例，无需额外依赖：

import csv
import sqlite3

def init_dedupe_db(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # 用主键约束天然实现去重
    cursor.execute('''CREATE TABLE IF NOT EXISTS seen_urls
                      (url TEXT PRIMARY KEY)''')
    conn.commit()
    conn.close()

def is_url_seen(db_path, url):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute('SELECT 1 FROM seen_urls WHERE url = ?', (url,))
    result = cursor.fetchone()
    conn.close()
    return result is not None

def mark_url_seen(db_path, url):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        cursor.execute('INSERT INTO seen_urls (url) VALUES (?)', (url,))
        conn.commit()
    except sqlite3.IntegrityError:
        # URL已存在，直接忽略
        pass
    finally:
        conn.close()

def export_backlinks_streaming(backlink_generator, output_file, dedupe_db):
    init_dedupe_db(dedupe_db)
    with open(output_file, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["url", "anchor_text", "authority"])
        
        # 遍历生成器，边获取边处理，不占用大量内存
        for link in backlink_generator:
            url = link["url"]
            if not is_url_seen(dedupe_db, url):
                mark_url_seen(dedupe_db, url)
                writer.writerow([url, link["anchor_text"], link["authority"]])

二、处理分页与速率限制 #

不管是调用第三方API还是爬取数据，分页和速率限制都是必处理的环节。下面是一个内置重试、限速逻辑的分页数据获取生成器：

import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session():
    session = requests.Session()
    # 配置自动重试规则，针对429、5xx等错误
    retry = Retry(
        total=5,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount('https://', adapter)
    session.mount('http://', adapter)
    return session

def get_backlinks_with_pagination(api_base_url, domain, api_key, page_size=100):
    session = create_retry_session()
    page = 1
    while True:
        params = {
            'domain': domain,
            'page': page,
            'page_size': page_size,
            'api_key': api_key
        }
        try:
            response = session.get(api_base_url, params=params)
            response.raise_for_status()
            data = response.json()
            backlinks = data.get('backlinks', [])
            if not backlinks:
                break  # 无更多数据，终止循环
            
            yield from backlinks  # 流式返回当前页数据
            
            # 根据API返回的限速头处理等待逻辑
            remaining = int(response.headers.get('X-RateLimit-Remaining', 1))
            if remaining <= 1:
                reset_time = int(response.headers.get('X-RateLimit-Reset', time.time() + 60))
                sleep_time = max(reset_time - time.time(), 1)
                time.sleep(sleep_time)
            
            page += 1
        except requests.exceptions.RequestException as e:
            print(f"获取第{page}页失败: {str(e)}")
            time.sleep(5)  # 出错后等待重试

三、多来源数据合并处理 #

多个数据源的合并可以直接通过生成器链式处理，完全无需加载全量数据到内存：

from itertools import chain

def merge_multiple_sources(source_generators):
    # 合并多个流式数据源
    yield from chain(*source_generators)

# 使用示例
source1 = get_backlinks_with_pagination('https://api.source1.com/backlinks', 'example.com', 'your_key1')
source2 = get_backlinks_with_pagination('https://api.source2.com/links', 'example.com', 'your_key2')
merged_backlinks = merge_multiple_sources([source1, source2])

# 调用流式导出函数
export_backlinks_streaming(merged_backlinks, 'final_backlinks.csv', 'dedupe_record.db')

额外优化建议 #

• 若无需长期保留去重记录，可改用布隆过滤器替代SQLite，内存占用更低，仅存在极小误判率，适合对去重精度要求非100%的场景。
• 写入CSV时可改用csv.DictWriter，直接传入字典字段，简化代码。
• 处理百万级以上超大规模数据时，可改用LevelDB等高效磁盘存储，或拆分生成多个CSV文件分块处理。

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