脚本之外:为什么建筑问题
一个单文件的刮纸机可以做一百页。 但当你需要收集数十个目标站点的数百万个数据点时,你需要一个系统而不是脚本. 可靠的刮刮结构将关切分为独立的组件,这些组件可以独立扩大、监测和回收。
本指南贯穿生产刮刮系统的设计,从URL调度到代理管理到数据存储. 每个组件都用代码来显示,并与 代理服务器基础设施。 。 。
建筑精良的刮刮机将数据收集视为工程问题,而不是黑客问题. 每个组件都有单一的责任,清晰的界面,以及可观察到的行为.
系统架构概览
生产刮刮系统由六个核心部分组成:
┌─────────────┐ ┌──────────────┐ ┌─────────────────┐
│ Scheduler │────▶│ URL Queue │────▶│ Crawler Pool │
│ (cron/API) │ │ (Redis) │ │ (workers) │
└─────────────┘ └──────────────┘ └────────┬────────┘
│
┌────────▼────────┐
│ Proxy Layer │
│ (ProxyHat) │
└────────┬────────┘
│
┌────────▼────────┐
│ Parser │
│ (extract data) │
└────────┬────────┘
│
┌──────────────┐ ┌────────▼────────┐
│ Monitoring │◀─────│ Storage │
│ (metrics) │ │ (DB / files) │
└──────────────┘ └─────────────────┘
| 构成部分 | 责任 | 技术 |
|---|---|---|
| 调度器 | 决定刮什么和何时刮什么 | Cron, Celery Beat, 公牛队 |
| URL 队列 | 具有优先级和减号的缓冲 URL | 雷迪斯, 兔子MQ, SQS |
| 爬行池 | 同步获取页面 | Ayncio,goroutines,工人线条 |
| 代理图层 | 通过旋转代理请求路由 | 代理用户端 |
| 解析器 | 从 HTML/JSON 提取结构化数据 | 美丽的苏普 欢乐 欢乐 |
| 储存 | 持久性提取数据 | PostgreSQL, 蒙戈DB, S3 |
| 监测 | 跟踪健康和业绩 | 普罗米修斯、伐木、警报 |
构成部分1:调度器
调度器决定要爬行的 URL 和 时间 。 它管理爬行频率,优先级,并确保没有URL被刮掉的次数超过必要的次数.
import redis
import json
import time
from datetime import datetime, timedelta
class CrawlScheduler:
"""Manages crawl schedules and feeds URLs to the queue."""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis = redis.from_url(redis_url)
self.queue_key = "scraper:url_queue"
self.schedule_key = "scraper:schedules"
self.seen_key = "scraper:seen_urls"
def add_schedule(self, name: str, urls: list[str], interval_hours: int, priority: int = 5):
"""Register a recurring crawl job."""
self.redis.hset(self.schedule_key, name, json.dumps({
"urls": urls,
"interval_hours": interval_hours,
"priority": priority,
"last_run": None,
}))
def tick(self):
"""Check all schedules and enqueue URLs that are due."""
schedules = self.redis.hgetall(self.schedule_key)
enqueued = 0
for name, data in schedules.items():
schedule = json.loads(data)
last_run = schedule.get("last_run")
interval = timedelta(hours=schedule["interval_hours"])
if last_run and datetime.fromisoformat(last_run) + interval > datetime.utcnow():
continue
for url in schedule["urls"]:
self.enqueue(url, priority=schedule["priority"])
enqueued += 1
schedule["last_run"] = datetime.utcnow().isoformat()
self.redis.hset(self.schedule_key, name, json.dumps(schedule))
return enqueued
def enqueue(self, url: str, priority: int = 5):
"""Add a URL to the crawl queue with deduplication."""
# Skip if recently seen (within 1 hour)
if self.redis.sismember(self.seen_key, url):
return False
self.redis.zadd(self.queue_key, {
json.dumps({"url": url, "enqueued_at": time.time()}): priority
})
self.redis.sadd(self.seen_key, url)
self.redis.expire(self.seen_key, 3600) # 1-hour dedup window
return True
def dequeue(self, batch_size: int = 10) -> list[dict]:
"""Pull the highest-priority URLs from the queue."""
items = self.redis.zpopmax(self.queue_key, batch_size)
return [json.loads(item) for item, score in items]
@property
def queue_size(self) -> int:
return self.redis.zcard(self.queue_key)
# Usage
scheduler = CrawlScheduler()
scheduler.add_schedule(
name="product_prices",
urls=[f"https://example.com/product/{i}" for i in range(1, 1001)],
interval_hours=6,
priority=8,
)
scheduler.add_schedule(
name="competitor_pages",
urls=["https://competitor.com/pricing", "https://competitor.com/features"],
interval_hours=24,
priority=5,
)
# Run every minute via cron
enqueued = scheduler.tick()
print(f"Enqueued {enqueued} URLs, queue size: {scheduler.queue_size}")
构成部分2: URL 队列
排队脱离了爬行的排队 它提供优先排序,反压,和持久性——所以如果爬行者崩溃,不会丢失URL.
// Node.js queue with Bull
const Queue = require('bull');
const crawlQueue = new Queue('crawl', 'redis://localhost:6379', {
defaultJobOptions: {
attempts: 3,
backoff: { type: 'exponential', delay: 2000 },
removeOnComplete: 100,
removeOnFail: 500,
},
});
// Add URLs with priority (lower number = higher priority)
async function enqueueUrls(urls, priority = 5) {
const jobs = urls.map(url =>
crawlQueue.add(
{ url, enqueuedAt: Date.now() },
{ priority, jobId: url } // jobId for deduplication
)
);
await Promise.all(jobs);
console.log(`Enqueued ${urls.length} URLs`);
}
// Worker processes URLs
crawlQueue.process(10, async (job) => {
const { url } = job.data;
// Crawl logic here (see Crawler Pool below)
return { url, status: 'completed' };
});
crawlQueue.on('failed', (job, err) => {
console.error(`Job ${job.id} failed: ${err.message}`);
});
构成部分3:爬行池
爬行池会同时通过 代理软件中间层它管理 货币限额,处理复试,并将原始回复传递给解析器。
import asyncio
import aiohttp
import uuid
import time
from typing import Optional
class CrawlerPool:
"""Concurrent crawler with proxy rotation and retry logic."""
def __init__(
self,
concurrency: int = 30,
max_retries: int = 3,
timeout: int = 30,
):
self.concurrency = concurrency
self.max_retries = max_retries
self.timeout = aiohttp.ClientTimeout(total=timeout)
self.semaphore = asyncio.Semaphore(concurrency)
self.stats = {"success": 0, "failed": 0, "retries": 0}
def _get_proxy(self, country: Optional[str] = None) -> str:
session_id = uuid.uuid4().hex[:8]
username = f"USERNAME-session-{session_id}"
if country:
username += f"-country-{country}"
return f"http://{username}:PASSWORD@gate.proxyhat.com:8080"
async def fetch(
self,
session: aiohttp.ClientSession,
url: str,
country: Optional[str] = None,
) -> dict:
async with self.semaphore:
for attempt in range(self.max_retries + 1):
proxy = self._get_proxy(country)
start = time.time()
try:
async with session.get(
url, proxy=proxy, timeout=self.timeout,
headers={
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/131.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,*/*;q=0.8",
},
) as response:
body = await response.text()
latency = time.time() - start
if response.status in (403, 429, 503) and attempt < self.max_retries:
self.stats["retries"] += 1
await asyncio.sleep(2 ** attempt)
continue
if response.status < 400:
self.stats["success"] += 1
else:
self.stats["failed"] += 1
return {
"url": url,
"status": response.status,
"body": body,
"latency": latency,
"success": response.status < 400,
}
except Exception as e:
if attempt < self.max_retries:
self.stats["retries"] += 1
await asyncio.sleep(2 ** attempt)
continue
self.stats["failed"] += 1
return {
"url": url,
"error": str(e),
"latency": time.time() - start,
"success": False,
}
async def crawl(self, urls: list[str], country: Optional[str] = None) -> list[dict]:
async with aiohttp.ClientSession() as session:
tasks = [self.fetch(session, url, country) for url in urls]
return await asyncio.gather(*tasks)
# Usage
crawler = CrawlerPool(concurrency=30, max_retries=3)
urls = [f"https://example.com/product/{i}" for i in range(500)]
results = asyncio.run(crawler.crawl(urls, country="us"))
print(f"Success: {crawler.stats['success']}, Failed: {crawler.stats['failed']}")
构成部分4:解析器
解析器将原始 HTML 转换为结构化的数据. 保持解析逻辑与爬行分离——这使得它可以测试,可重复使用,在目标站点改变时容易更新.
from dataclasses import dataclass
from typing import Optional
from bs4 import BeautifulSoup
import json
@dataclass
class ProductData:
url: str
title: Optional[str] = None
price: Optional[float] = None
currency: Optional[str] = None
availability: Optional[str] = None
rating: Optional[float] = None
review_count: Optional[int] = None
parsed_at: Optional[str] = None
class ProductParser:
"""Extracts structured product data from HTML."""
def parse(self, url: str, html: str) -> ProductData:
soup = BeautifulSoup(html, "html.parser")
data = ProductData(url=url)
try:
data.title = self._extract_text(soup, "h1.product-title")
data.price = self._extract_price(soup)
data.currency = self._extract_currency(soup)
data.availability = self._extract_text(soup, ".availability-status")
data.rating = self._extract_rating(soup)
data.review_count = self._extract_review_count(soup)
data.parsed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
except Exception as e:
logger.error(f"Parse error for {url}: {e}")
return data
def _extract_text(self, soup, selector: str) -> Optional[str]:
el = soup.select_one(selector)
return el.get_text(strip=True) if el else None
def _extract_price(self, soup) -> Optional[float]:
el = soup.select_one("[data-price], .price, .product-price")
if not el:
return None
price_text = el.get("data-price") or el.get_text(strip=True)
# Remove currency symbols and parse
cleaned = "".join(c for c in price_text if c.isdigit() or c == ".")
return float(cleaned) if cleaned else None
def _extract_currency(self, soup) -> Optional[str]:
el = soup.select_one("[data-currency], .currency")
return el.get("data-currency") or el.get_text(strip=True) if el else None
def _extract_rating(self, soup) -> Optional[float]:
el = soup.select_one("[data-rating], .rating-value")
if el:
val = el.get("data-rating") or el.get_text(strip=True)
try:
return float(val)
except ValueError:
return None
return None
def _extract_review_count(self, soup) -> Optional[int]:
el = soup.select_one(".review-count, [data-reviews]")
if el:
text = el.get("data-reviews") or el.get_text(strip=True)
digits = "".join(c for c in text if c.isdigit())
return int(digits) if digits else None
return None
# Usage
parser = ProductParser()
for result in results:
if result["success"]:
product = parser.parse(result["url"], result["body"])
print(f"{product.title}: ${product.price}")
构成部分5:储存
坚持用版本和调值解析数据. 使用升降器来优雅地处理重爬行.
import psycopg2
from psycopg2.extras import execute_values
from datetime import datetime
class DataStore:
"""Persists scraped data with upsert and versioning."""
def __init__(self, dsn: str):
self.conn = psycopg2.connect(dsn)
self._ensure_tables()
def _ensure_tables(self):
with self.conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS products (
url TEXT PRIMARY KEY,
title TEXT,
price NUMERIC(10,2),
currency VARCHAR(3),
availability TEXT,
rating NUMERIC(3,2),
review_count INTEGER,
first_seen_at TIMESTAMPTZ DEFAULT NOW(),
last_updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS price_history (
id SERIAL PRIMARY KEY,
url TEXT REFERENCES products(url),
price NUMERIC(10,2),
recorded_at TIMESTAMPTZ DEFAULT NOW()
);
""")
self.conn.commit()
def upsert_products(self, products: list[ProductData]):
"""Insert or update products, recording price changes."""
with self.conn.cursor() as cur:
for product in products:
if product.price is None:
continue
# Check if price changed
cur.execute(
"SELECT price FROM products WHERE url = %s",
(product.url,)
)
row = cur.fetchone()
# Upsert product
cur.execute("""
INSERT INTO products (url, title, price, currency, availability,
rating, review_count, last_updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (url) DO UPDATE SET
title = EXCLUDED.title,
price = EXCLUDED.price,
currency = EXCLUDED.currency,
availability = EXCLUDED.availability,
rating = EXCLUDED.rating,
review_count = EXCLUDED.review_count,
last_updated_at = NOW()
""", (
product.url, product.title, product.price,
product.currency, product.availability,
product.rating, product.review_count,
))
# Record price history if changed
if row is None or float(row[0]) != product.price:
cur.execute(
"INSERT INTO price_history (url, price) VALUES (%s, %s)",
(product.url, product.price)
)
self.conn.commit()
@property
def product_count(self) -> int:
with self.conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM products")
return cur.fetchone()[0]
# Usage
store = DataStore("postgresql://user:pass@localhost:5432/scraper")
store.upsert_products(parsed_products)
print(f"Total products: {store.product_count}")
构成部分6:监测
将所有部件与集中监测联系起来。 具体代用监测模式见我们 监视代理性能 向导。
import time
import json
import logging
from datetime import datetime
class PipelineMonitor:
"""Monitors the entire scraping pipeline."""
def __init__(self):
self.logger = logging.getLogger("pipeline")
self.stage_times = {}
self.stage_counts = {}
def start_stage(self, stage: str):
self.stage_times[stage] = time.time()
def end_stage(self, stage: str, item_count: int = 0):
elapsed = time.time() - self.stage_times.get(stage, time.time())
self.stage_counts[stage] = item_count
self.logger.info(json.dumps({
"timestamp": datetime.utcnow().isoformat(),
"stage": stage,
"duration_seconds": round(elapsed, 2),
"items_processed": item_count,
"items_per_second": round(item_count / elapsed, 1) if elapsed > 0 else 0,
}))
def report(self) -> dict:
return {
"stages": {
stage: {
"items": self.stage_counts.get(stage, 0),
}
for stage in self.stage_counts
},
"timestamp": datetime.utcnow().isoformat(),
}
# Usage within the full pipeline
monitor = PipelineMonitor()
# Stage 1: Schedule
monitor.start_stage("schedule")
urls = scheduler.dequeue(batch_size=500)
monitor.end_stage("schedule", len(urls))
# Stage 2: Crawl
monitor.start_stage("crawl")
results = asyncio.run(crawler.crawl([u["url"] for u in urls]))
monitor.end_stage("crawl", len(results))
# Stage 3: Parse
monitor.start_stage("parse")
products = [parser.parse(r["url"], r["body"]) for r in results if r["success"]]
monitor.end_stage("parse", len(products))
# Stage 4: Store
monitor.start_stage("store")
store.upsert_products(products)
monitor.end_stage("store", len(products))
print(json.dumps(monitor.report(), indent=2))
将它放在一起
以下是连接所有六个组件的完整管道,连接到一个单一的,可循环的刮刮系统.
import asyncio
import logging
logging.basicConfig(level=logging.INFO)
async def run_pipeline():
# Initialize components
scheduler = CrawlScheduler()
crawler = CrawlerPool(concurrency=30, max_retries=3)
parser = ProductParser()
store = DataStore("postgresql://user:pass@localhost:5432/scraper")
monitor = PipelineMonitor()
# Schedule crawls
scheduler.add_schedule(
name="daily_products",
urls=[f"https://example.com/product/{i}" for i in range(1, 501)],
interval_hours=24,
priority=8,
)
# Main loop
while True:
# 1. Check schedules and enqueue URLs
monitor.start_stage("schedule")
scheduler.tick()
batch = scheduler.dequeue(batch_size=100)
monitor.end_stage("schedule", len(batch))
if not batch:
await asyncio.sleep(60)
continue
# 2. Crawl
monitor.start_stage("crawl")
urls = [item["url"] for item in batch]
results = await crawler.crawl(urls, country="us")
successful = [r for r in results if r.get("success")]
monitor.end_stage("crawl", len(successful))
# 3. Parse
monitor.start_stage("parse")
products = []
for result in successful:
try:
product = parser.parse(result["url"], result["body"])
products.append(product)
except Exception as e:
logging.error(f"Parse error: {e}")
monitor.end_stage("parse", len(products))
# 4. Store
monitor.start_stage("store")
store.upsert_products(products)
monitor.end_stage("store", len(products))
# 5. Report
logging.info(f"Pipeline: {monitor.report()}")
logging.info(f"Crawler stats: {crawler.stats}")
# Wait before next batch
await asyncio.sleep(5)
asyncio.run(run_pipeline())
生产部署模式
水平缩放
缩放爬行器独立于排程器和剖析器. 运行从相同的 Redis 队列中消耗的多个爬行例 。
| 构成部分 | 扩大战略 | 典型缩放 |
|---|---|---|
| 调度器 | 单一案件(领袖选举) | 1个实例 |
| URL 队列 | 重新编组或管理队列 | 1组 |
| 爬行池 | 水平吊舱缩放 | 2-20个案例 |
| 代理图层 | 由代理哈特公司管理 | 不详(外部) |
| 解析器 | 与爬行者或分离者合用同一地点 | 1: 1 有爬行器 |
| 储存 | 数据库复制 | 1个主要复制品 |
| 监测 | 集中汇总 | 1个实例 |
恢复错误
- Crawler 崩溃 : URL仍然在 Redis 队列中 。 新的爬行者会自动捡起它们.
- 代理失败 : 这个 中间软件层 与新鲜IP重试. 持续失败引发警报.
- 解析器失败 : Raw HTML 存储在一个死字母队列中,用于手动检查和解析更新.
- 数据库失败 : 在内存/磁盘中用写头记录解析数据缓冲器,直到 DB 恢复.
数据质量检查
class DataQualityChecker:
"""Validates parsed data before storage."""
def check(self, product: ProductData) -> list[str]:
issues = []
if not product.title:
issues.append("missing_title")
if product.price is not None and product.price <= 0:
issues.append("invalid_price")
if product.price is not None and product.price > 100000:
issues.append("suspicious_price")
if product.rating is not None and (product.rating < 0 or product.rating > 5):
issues.append("invalid_rating")
return issues
def filter_valid(self, products: list[ProductData]) -> list[ProductData]:
valid = []
for product in products:
issues = self.check(product)
if issues:
logger.warning(f"Data quality issues for {product.url}: {issues}")
else:
valid.append(product)
return valid
# Integrate into pipeline
checker = DataQualityChecker()
valid_products = checker.filter_valid(products)
store.upsert_products(valid_products)
最好的刮刮建筑 就是你能解释的建筑 每个部分应回答三个问题: 这有什么用? 怎么会失败? 怎样恢复?
关于代理层组件,见 构建代理软件中层。将优化爬行器吞吐量改为 以货币控制放大代理请求。关于基于浏览器的爬行,参见 与 Playwright 旋转代理对于反侦查战略,探索 如何不受阻地刮刮。 。 。
开始 Python SDK 键盘, (中文). 节点 SDK,或 冲啊 SDK 用于代理整合。 探索 代理用户定价 和 文档 来为你的刮刮建筑提供动力
