python scrapy爬虫分布式部署

熊黎龙2022年5月24日大约 3 分钟约 923 字

分布式`Scrapy`爬虫

[TOC]

开始

scheduler连接到Redis把Request队列和URL去重数据序列化放入了Redis

pipeline连接到Redis把Item数据序列化放入了Redis

`scrapy-redis`安装

安装redis驱动：pip install redis

克隆scrapy-redis包：scrapy-redis

将scrapy-redis/src/scrapy_redis这个文件夹拷贝到scrapy项目的根目录

`scrapy-redis`配置

项目基本配置

>>> vim ArticleSpider/settings.py

# 在redis中启用调度存储请求队列
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 确保所有蜘蛛通过redis共享相同的重复过滤器.
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# 将已删除的项目存储在redis中进行后期处理。
ITEM_PIPELINES  = {
     'scrapy_redis.pipelines.RedisPipeline':300
}

爬虫模板

>>> vim ArticleSpider/spiders/jobbole.py

from scrapy_redis.spiders import RedisSpider

class JobboleSpider(RedisSpider):
    name = 'jobbole'
    allowed_domains = ['blog.jobbole.com']
    # start_urls = ['http://blog.jobbole.com/']  # 没有必要了

    def parse(self, response):
        pass

启动爬虫后，爬虫将会一直监听Redis数据库中的数据，不会停止。

开始爬取

PS C:\Program Files\Redis> .\redis-cli.exe
127.0.0.1:6379> lpush jobbole:start_urls http://blog.jobbole.com/all-posts/

源码

源文件	内容
`connection.py`	`Redis`连接文件
`defaults.py`	配置文件
`dupefilter.py`	替换`Scrapy`默认的URL去重器
`picklecompat.py`	对象的序列化和反序列化
`pipelines.py`	将Item保存到`Redis`
`queue.py`	在`Redis`中实现了三种Request队列
`scheduler.py`	连接`Redis`的调度器
`utils.py`	python3的兼容

`defaults.py`

>>> vim scrapy_redis/defaults.py

import redis

# 保存每一个访问过的Request的指纹
DUPEFILTER_KEY = 'dupefilter:%(timestamp)s'

# 每一个Spider的Item队列
PIPELINE_KEY = '%(spider)s:items'

# 使用redis-py驱动
REDIS_CLS = redis.StrictRedis
# 连接Redis的编码
REDIS_ENCODING = 'utf-8'
# redis-py连接参数
REDIS_PARAMS = {
    'socket_timeout': 30,
    'socket_connect_timeout': 30,
    'retry_on_timeout': True,
    'encoding': REDIS_ENCODING,
}

# Request队列的Redis变量名
SCHEDULER_QUEUE_KEY = '%(spider)s:requests'

# Request队列的类型
SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue'

# 去重过滤器的Redis变量名
SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter'

# 去重过滤器的类型
SCHEDULER_DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter'

# 开始URL的Redis变量名
START_URLS_KEY = '%(name)s:start_urls'
START_URLS_AS_SET = False

集成`bloomfilter`到`scrapy-redis`中

`bloomfilter源码`

直接拷贝到项目中使用

>>> vim ArticleSpider/utils/bloomfilter.py

import mmh3  # pip install mmh3
import redis
import math


class PyBloomFilter():
    # 内置100个随机种子
    SEEDS = [543, 460, 171, 876, 796, 607, 650, 81, 837, 545, 591, 946, 846, 521, 913, 636, 878, 735, 414, 372, 344, 324, 223, 180, 327, 891, 798, 933, 493, 293, 836, 10, 6, 544, 924, 849, 438, 41, 862, 648, 338, 465, 562, 693, 979, 52, 763, 103, 387, 374, 349, 94, 384, 680, 574, 480, 307, 580, 71, 535, 300, 53, 481, 519, 644, 219, 686, 236, 424, 326, 244, 212, 909, 202, 951, 56, 812, 901, 926, 250, 507, 739, 371, 63, 584, 154, 7, 284, 617, 332, 472, 140, 605, 262, 355, 526, 647, 923, 199, 518]

    # capacity是预先估计要去重的数量
    # error_rate表示错误率
    # conn表示redis的连接客户端
    # key表示在redis中的键的名字前缀
    def __init__(self,capacity=1000000000,error_rate=0.00000001,conn=None,key='BloomFilter'):
        # 需要的总bit位数
        self.m = math.ceil(capacity*math.log2(math.e)*math.log2(1/error_rate)) 
        # 需要最少的hash次数
        self.k = math.ceil(math.log1p(2)*self.m/capacity)   
        # 需要的多少M内存
        self.mem = math.ceil(self.m/8/1024/1024)     
        # 需要多少个512M的内存块,value的第一个字符必须是ascii码，所有最多有256个内存块
        self.blocknum = math.ceil(self.mem/512)                                 
        self.seeds = self.SEEDS[0:self.k]
        self.key = key
        self.N = 2**31-1
        self.redis = conn
        #print(self.mem)
        #print(self.k)

    def add(self, value):
        name = self.key + "_" + str(ord(value[0])%self.blocknum)
        hashs = self.get_hashs(value)
        for hash in hashs:
            self.redis.setbit(name, hash, 1)

    def is_exist(self, value):
        name = self.key + "_" + str(ord(value[0])%self.blocknum)
        hashs = self.get_hashs(value)
        exist = True
        for hash in hashs:
            exist = exist & self.redis.getbit(name, hash)
        return exist

    def get_hashs(self, value):
        hashs = list()
        for seed in self.seeds:
            hash = mmh3.hash(value, seed)
            if hash >= 0:
                hashs.append(hash)
            else:
                hashs.append(self.N - hash)
        return hashs


pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=0)
conn = redis.StrictRedis(connection_pool=pool)


if __name__ == "__main__":
    bf = PyBloomFilter(conn=conn)
    bf.add('www.jobbole.com')
    bf.add('www.zhihu.com')
    print(bf.is_exist('www.zhihu.com'))
    print(bf.is_exist('www.lagou.com'))

集成到`scrapy_redis`的去重器`dupefilter.py`文件内

>>> vim scrapy_redis/dupefilter.py

from ArticleSpider.utils.bloomfilter import PyBloomFilter, conn
class RFPDupeFilter(BaseDupeFilter):
    logger = logger
    def __init__(self, server, key, debug=False):
        self.server = server
        self.key = key
        self.debug = debug
        self.logdupes = True
        
        self.bf = PyBloomFilter(conn=conn, key=key)  # 添加这个
        
    @classmethod
    def from_settings(cls, settings):
        pass
    
    # 重写这个方法
    def request_seen(self, request):
        fp = self.request_fingerprint(request)
        if self.bf.is_exist(fp):
            return True
        else:
            self.bf.add(fp)
            return False