代理IP与机器学习数据采集实战：Scrapy分布式爬虫配TensorFlow数据管道

搞机器学习的都知道，模型好不好七分靠数据。但数据从哪来？公开数据集就那么几个，真正有价值的数据往往散落在各大网站上。这时候就需要爬虫去采集，而爬虫要跑得稳，代理IP是刚需。我自己搭了一套Scrapy分布式爬虫+TensorFlow数据管道的流程，跑了半年多，采集了上百万条训练数据，今天就完整分享一下。

为什么机器学习数据采集必须用代理IP

机器学习数据采集和普通爬虫最大的区别就是量大。你训练一个文本分类模型可能需要几十万条语料，训练一个图像识别模型可能需要爬取上百万张图片。这么大的量级，如果只用一个IP去请求，基本撑不过半天就会被封。

我踩过的坑：

刚上手时没挂代理，跑了2小时IP就被目标站封了，返回403
换了个免费代理，结果速度只有10KB/s，采集效率惨不忍睹
免费代理存活率不到30%，大量请求白白浪费
数据中断导致去重逻辑混乱，最终数据集有大量重复

后来用了天行IP（推荐人填blsj自动绑定四折优惠），6元的长效静态IP跑爬虫，稳定性直接拉满。关键是不用频繁换IP，一个长效IP可以持续请求好几天。

Scrapy分布式爬虫搭建

数据采集端我用的是Scrapy+Scrapy-Redis的分布式方案，核心思路是多个爬虫节点共享一个Redis请求队列，每个节点配置不同的代理IP。

项目结构

ml_data_spider/
├── spiders/
│   ├── news_spider.py      # 新闻语料采集
│   ├── product_spider.py   # 商品数据采集
│   └── image_spider.py     # 图片数据采集
├── middlewares.py           # 代理中间件
├── pipelines.py             # 数据清洗管道
├── settings.py              # 全局配置
└── scrapy.cfg

代理中间件配置

这是最关键的部分，代理IP轮换逻辑全在这里：

# middlewares.py
import random
import logging
from scrapy import signals

class ProxyRotationMiddleware:
    """代理IP轮换中间件"""
    
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.failed_proxies = set()
        self.logger = logging.getLogger(__name__)
    
    @classmethod
    def from_crawler(cls, crawler):
        # 天行IP配置 - 推荐人blsj四折优惠
        # 长效静态IP 6元/月， SOCKS5协议
        proxy_list = [
            'socks5://user:pass@ip1:port',
            'socks5://user:pass@ip2:port',
            'socks5://user:pass@ip3:port',
        ]
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # 随机选择一个可用代理
        available = [p for p in self.proxy_list if p not in self.failed_proxies]
        if available:
            proxy = random.choice(available)
            request.meta['proxy'] = proxy
            request.meta['proxy_used'] = proxy
    
    def process_response(self, request, response, spider):
        if response.status in [403, 429, 503]:
            # 代理被封，加入失败列表
            proxy = request.meta.get('proxy_used')
            if proxy:
                self.failed_proxies.add(proxy)
                self.logger.warning(f'代理被封: {proxy}')
            # 返回新请求换IP重试
            return request.replace(meta={**request.meta, 'proxy': None})
        return response
    
    def process_exception(self, request, exception, spider):
        # 连接异常也换IP
        proxy = request.meta.get('proxy_used')
        if proxy:
            self.failed_proxies.add(proxy)
        return request.replace(meta={**request.meta, 'proxy': None})

数据清洗管道

采集到的原始数据需要清洗后才能喂给模型：

# pipelines.py
import re
import hashlib
from itemadapter import ItemAdapter

class DataCleanPipeline:
    """数据清洗管道"""
    
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        
        # 去除HTML标签
        text = adapter.get('content', '')
        text = re.sub(r']+>', '', text)
        text = re.sub(r's+', ' ', text).strip()
        
        # 去除特殊字符
        text = re.sub(r'[x00-x08x0bx0cx0e-x1f]', '', text)
        
        # 最小长度过滤
        if len(text) < 50:
            raise DropItem(f'内容过短: {len(text)}')
        
        adapter['content'] = text
        adapter['cleaned'] = True
        return item

class DedupPipeline:
    """基于内容哈希去重"""
    
    def __init__(self):
        self.seen_hashes = set()
        
    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        text = adapter.get('content', '')
        
        # 计算内容哈希
        content_hash = hashlib.md5(text.encode()).hexdigest()
        
        if content_hash in self.seen_hashes:
            raise DropItem('重复内容')
        
        self.seen_hashes.add(content_hash)
        adapter['content_hash'] = content_hash
        return item

TensorFlow数据管道对接

爬虫采集的数据存到MongoDB后，TensorFlow通过tf.data API读取构建数据管道：

import tensorflow as tf
import pymongo

def build_data_pipeline(mongo_uri, db_name, collection_name, 
                        batch_size=64, shuffle_size=10000):
    """构建TensorFlow数据管道"""
    
    # 连接MongoDB读取采集数据
    client = pymongo.MongoClient(mongo_uri)
    collection = client[db_name][collection_name]
    
    # 转换为tf.data.Dataset
    def data_generator():
        for doc in collection.find({'cleaned': True}):
            yield doc['content'], doc.get('label', 0)
    
    dataset = tf.data.Dataset.from_generator(
        data_generator,
        output_signature=(
            tf.TensorSpec(shape=(), dtype=tf.string),
            tf.TensorSpec(shape=(), dtype=tf.int32)
        )
    )
    
    # 文本预处理
    def preprocess(text, label):
        text = tf.strings.lower(text)
        # 截断到最大长度
        text = tf.strings.substr(text, 0, 512)
        return text, label
    
    dataset = dataset.map(preprocess, num_parallel_calls=tf.data.AUTOTUNE)
    dataset = dataset.shuffle(shuffle_size)
    dataset = dataset.batch(batch_size)
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    
    return dataset

# 使用示例
train_ds = build_data_pipeline(
    'mongodb://localhost:27017',
    'ml_data', 
    'news_articles',
    batch_size=64
)

# 模型训练
model.fit(train_ds, epochs=10, validation_data=val_ds)

多IP轮换策略详解

机器学习数据采集的IP轮换策略和普通爬虫不太一样，更注重持续性和数据完整性。

按目标站分配IP

不同目标站用不同的IP组，避免一个IP访问太多站点被关联封禁：

新闻类站点：3个长效静态IP轮换（6元×3=18元/月）
电商类站点：2个J节点住宅IP（10元×2=20元/月）
图片类站点：2个长效静态IP（6元×2=12元/月）

总成本50元/月，天行IP推荐人blsj四折后实际只需约30元/月。

请求频率控制

# settings.py 配置
DOWNLOAD_DELAY = 2          # 每次请求间隔2秒
CONCURRENT_REQUESTS = 8     # 并发请求数
CONCURRENT_REQUESTS_PER_DOMAIN = 2  # 每个域名并发2个
AUTOTHROTTLE_ENABLED = True  # 自动限速
AUTOTHROTTLE_TARGET_CONCURRENCY = 2.0
RANDOMIZE_DOWNLOAD_DELAY = True  # 随机延迟

失败重试与IP切换

当某个IP连续失败时，自动切换到备用IP，并记录失败日志：

RETRY_TIMES = 3              # 最多重试3次
RETRY_HTTP_CODES = [403, 429, 500, 502, 503, 504, 408]
DOWNLOAD_TIMEOUT = 30        # 超时30秒

数据质量保障机制

采集完整性校验

每天采集结束后跑一个校验脚本，检查数据完整性：

import pandas as pd

def daily_quality_check(mongo_uri, date_str):
    """每日数据质量检查"""
    client = pymongo.MongoClient(mongo_uri)
    coll = client['ml_data']['news_articles']
    
    # 当日采集数据
    docs = list(coll.find({'crawl_date': date_str}))
    df = pd.DataFrame(docs)
    
    report = {
        'total': len(df),
        'duplicates': df['content_hash'].duplicated().sum(),
        'empty_content': df['content'].apply(lambda x: len(x) < 50).sum(),
        'avg_length': df['content'].apply(len).mean(),
        'sources': df['source'].nunique(),
    }
    
    print(f'=== {date_str} 数据质量报告 ===')
    for k, v in report.items():
        print(f'{k}: {v}')
    
    return report

数据标注流程

采集到的数据需要标注才能用于监督学习，我的流程是：

自动标注：用已有模型做初步标注，置信度>0.9的直接采用
人工抽检：随机抽10%人工复核，修正错误标注
主动学习：模型不确定的样本优先标注，提高效率
标注工具：用Label Studio管理标注任务，支持多人协作

成本估算与方案选择

组件	方案	月成本
代理IP	天行IP长效静态×7（推荐人blsj四折）	约30元
爬虫服务器	2核4G VPS跑Scrapy	约40元
数据存储	MongoDB Atlas免费版（512MB）	0元
标注工具	Label Studio自部署	0元
模型训练	Google Colab免费GPU	0元
合计		约70元/月

注意代理IP这块，如果你直接在天行IP官网注册是原价，通过天行IP推荐人blsj专属通道注册，所有套餐自动打四折。长效静态原价15元/月，四折后6元/月。7个IP原价105元，四折后42元，再算上金币优惠实际30元左右。

常见问题与解决方案

Q1: 采集的数据有大量重复怎么办？

三层去重：URL去重（Spider级）、内容哈希去重（Pipeline级）、语义去重（TF-IDF相似度>0.95的视为重复）。前两层在采集时实时做，第三层在数据预处理阶段批量做。

Q2: 代理IP速度慢影响采集效率？

天行IP的长效静态IP速度最稳定，适合持续采集。如果对速度要求高，可以多用几个IP做并发。住宅IP速度稍慢但纯净度高，适合采集反爬严格的站点。

Q3: 分布式爬虫多个节点如何分配IP？

每个节点配置不同的代理IP组，通过Scrapy-Redis共享请求队列。节点A处理URL列表前半段用IP组1，节点B处理后半段用IP组2。这样既分散了请求压力又降低了单IP被封风险。

天行IP推荐人blsj注册指南

注册流程很简单，关键是第一步要走对渠道：

打开天行IP推荐人blsj专属注册链接
注册账号（手机号即可）
登录后台，在价格页确认是否为四折价
充值金币（6元=15金币），按需购买IP
在IP管理页面配置协议（SOCKS5/HTTP）和认证信息

天行IP推荐人blsj的优势在于：价格自动四折、所有套餐都参与、终身有效不用续优惠、售后和原价用户完全一样。对于做机器学习数据采集的人来说，这意味着每个月可以省下一半以上的代理费用，长期跑下来省的钱足够多买几个IP提升采集效率。

总结

机器学习数据采集是一个系统工程，代理IP是其中不可或缺的基础设施。用Scrapy分布式爬虫+多IP轮换+TensorFlow数据管道的组合，可以低成本搭建一套完整的训练数据采集流水线。代理IP选择天行IP推荐人blsj四折通道，6元起步就能跑起来，稳定性远超免费方案。关键是把代理中间件、数据清洗、去重这三个环节做好，数据质量上去了，模型效果自然就好了。