Scrapy-redis 组件

Posted 2021-02-08 坨之歌

分布式爬虫

概念

所谓分布式, 多个程序同时对一个任务进行操作

一分多的高效率的任务进行方式

简单说明

一个 10GB 的爬虫任务, 交给10台服务器进行同时爬取

对比单服务器无论怎么优化都是 10倍的效率, 但是成本高

需要硬件环境支持 ( 带宽, 服务器设备等 )

多态主机共享一个爬取队列即为分布式爬虫

物理拓扑

 / -------------服务器 2

|　/ ------------------服务器3

服务器1    ----------- 服务器 4

|　\\-------------------服务器5

 \\-------------服务器6

服务器 1 负责 url队列的分发 ( 本质上是个 redis 数据库 )

其他服务器可能分布在全国各地, 分别和服务器1 建立联系,

通过 服务器 1 的共享队列进行爬取任务的获取

为什么使用 redis

redis 速度快 

redis 非关系型数据库, redis 中的集合,  存储每个 request 的指纹

Scrapy 实现分布式

Scrapy 原生是不支持分布式爬虫

Scrapy 的爬取队列是放在调度器中的

因此只要能够实现调度器的共享即可完成爬虫任务的共享

即重写 scrapy 的调度器, ( 别人造的轮子又大又圆 scrapy_redis )

scrapy-redis

简介

scrapy-redis是scrapy框架基于redis数据库的组件，用于scrapy项目的分布式开发和部署。

特征

分布式爬取

可以启动多个spider工程，相互之间共享单个redis的requests队列。最适合广泛的多个域名网站的内容爬取。

分布式数据处理

爬取到的scrapy的item数据可以推入到redis队列中

这意味着你可以根据需求启动尽可能多的处理程序来共享item的队列，进行item数据持久化处理

Scrapy即插即用组件

Scheduler调度器 + Duplication复制 过滤器，Item Pipeline，基本spider

三大作用

去重规则的校验

实现调度器对请求的调配

定制起始URL

scrapy-redis架构

安装

pip install scrapy-redis

使用

所有可选的配置, 未注释表示必选设置

# 替换调度器, 启用 redis 中存储的请求调度队列 
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 替换过滤器, 所有的通向 redis 的爬虫都需要使用此过滤器 ( 去重机制 )
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# 默认的 序列化方式为 pickle, 可以更改为 json 或者 msgpack
# SCHEDULER_SERIALIZER = "scrapy_redis.picklecompat"

# 是否云讯断点续爬 ( 爬取完成后是否清楚指纹 )
# SCHEDULER_PERSIST = True

# 默认使用优先级队列（默认），其他：PriorityQueue（有序集合），FifoQueue（列表）、LifoQueue（列表）
# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.PriorityQueue\'
# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.FifoQueue\'  # 广度优先 基于队列 先进先出
# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.LifoQueue\' # 深度优先 基于栈 后进先出


# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.PriorityQueue\'  的时候才有用
# DEPTH_PRIORITY = 1  # 广度优先
# DEPTH_PRIORITY = -1 # 深度优先 后进先出

# 初次启动的时候的阻塞时间 ( 仅适用于队列类为SpiderQueue或SpiderStack时 )
# SCHEDULER_IDLE_BEFORE_CLOSE = 10

# 替换持久化工具为 scrapy-redis 的
ITEM_PIPELINES = {
    \'scrapy_redis.pipelines.RedisPipeline\': 300
}

# The item pipeline serializes and stores the items in this redis key.
# REDIS_ITEMS_KEY = \'%(spider)s:items\'

# The items serializer is by default ScrapyJSONEncoder. You can use any
# importable path to a callable object.
# REDIS_ITEMS_SERIALIZER = \'json.dumps\'

# 指定连接 redis 的主机和端口号
# REDIS_HOST = \'localhost\'
# REDIS_PORT = 6379

# Specify the full Redis URL for connecting (optional).
# If set, this takes precedence over the REDIS_HOST and REDIS_PORT settings.
# REDIS_URL = \'redis://user:pass@hostname:9001\'

# Custom redis client parameters (i.e.: socket timeout, etc.)
# REDIS_PARAMS  = {}
# Use custom redis client class.
# REDIS_PARAMS[\'redis_cls\'] = \'myproject.RedisClient\'

# If True, it uses redis\' ``SPOP`` operation. You have to use the ``SADD``
# command to add URLs to the redis queue. This could be useful if you
# want to avoid duplicates in your start urls list and the order of
# processing does not matter.
# REDIS_START_URLS_AS_SET = False

# Default start urls key for RedisSpider and RedisCrawlSpider.
# REDIS_START_URLS_KEY = \'%(name)s:start_urls\'

# Use other encoding than utf-8 for redis.
# REDIS_ENCODING = \'latin1\'

去重

scrapy-redis  去重原理

 redis的集合有去重属性，在保存重复的值的时候返回值会是 0 

完全自定义实现去重

from scrapy.dupefilter import BaseDupeFilter
import redis
from scrapy.utils.request import request_fingerprint

class DupFilter(BaseDupeFilter):
    def __init__(self):
        self.conn = redis.Redis(host=\'140.143.227.206\',port=8888,password=\'beta\')

    def request_seen(self, request):
        """
        检测当前请求是否已经被访问过
        :param request: 
        :return: True表示已经访问过；False表示未访问过
        """
        fid = request_fingerprint(request)
        result = self.conn.sadd(\'visited_urls\', fid)
        if result == 1:
            return False
        return True

 使用 scrapy-redis 去重

# 直接加了配置就生效。但是是基于时间戳的，时间戳变了就会失效很不实用 

################ scrapy redis连接 ####################

REDIS_HOST = \'127.0.0.1\'                      　　　 # 主机名
REDIS_PORT = 6379                                   # 端口
REDIS_PARAMS  = {\'password\':\'yangtuo\'}              # Redis连接参数，比如密码              
# 默认：REDIS_PARAMS = {\'socket_timeout\': 30,\'socket_connect_timeout\': 30,\'retry_on_timeout\': True,\'encoding\': REDIS_ENCODING,}）
REDIS_ENCODING = "utf-8"                            # redis编码类型             默认：\'utf-8\'

# REDIS_URL = \'redis://user:pass@hostname:9001\'     # 连接URL（优先于以上配置）


################ scrapy redis去重 ####################

DUPEFILTER_KEY = \'dupefilter:%(timestamp)s\'    
# 默认的是按照时间戳的。结果每次请求的时候是会进行去重
# 但是下一次请求时间戳不一致就会清空导致无法去重 因此这里存在优化空间 最好定死

# DUPEFILTER_CLASS = \'scrapy_redis.dupefilter.RFPDupeFilter\'    # 如果使用 scrapy-redis 的默认去重是会使用这个类来处理去重 
DUPEFILTER_CLASS = \'dbd.xxx.RedisDupeFilter\' # 这样使用自己自定义的去重规则 

优化

from scrapy_redis.dupefilter import RFPDupeFilter
from scrapy_redis.connection import get_redis_from_settings
from scrapy_redis import defaults

class RedisDupeFilter(RFPDupeFilter):
    @classmethod
    def from_settings(cls, settings):

        server = get_redis_from_settings(settings)
        # key = defaults.DUPEFILTER_KEY % {\'timestamp\': int(time.time())}    # 源码这里是用的时间戳 不断变换导致每次重置
        key = defaults.DUPEFILTER_KEY % {\'timestamp\': \'yangtuo\'}    # 不要时间戳了，我定死一个固定值
        debug = settings.getbool(\'DUPEFILTER_DEBUG\')
        return cls(server, key=key, debug=debug)

调度器

在 settings.py 中进行配置 

SCHEDULER = "scrapy_redis.scheduler.Scheduler"


# 默认使用优先级队列（默认），其他：PriorityQueue（有序集合），FifoQueue（列表）、LifoQueue（列表）
SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.PriorityQueue\'  
# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.FifoQueue\'  # 广度优先 基于队列 先进先出 
# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.LifoQueue\' # 深度优先 基于栈 后进先出 


# SCHEDULER_QUEUE_CLASS = \'scrapy_redis.queue.PriorityQueue\'  的时候才有用 
DEPTH_PRIORITY = 1  # 广度优先    
# DEPTH_PRIORITY = -1 # 深度优先 后进先出


SCHEDULER_QUEUE_KEY = \'%(spider)s:requests\'  # 调度器中请求存放在redis中的key


SCHEDULER_SERIALIZER = "scrapy_redis.picklecompat"  # 对保存到redis中的数据进行序列化，默认使用pickle


SCHEDULER_PERSIST = False  # 是否在关闭时候保留原来的调度器和去重记录，True=保留，False=清空
SCHEDULER_FLUSH_ON_START = True  # 是否在开始之前清空 调度器和去重记录，True=清空，False=不清空
# SCHEDULER_IDLE_BEFORE_CLOSE = 10  # 去调度器中获取数据时，如果为空，最多等待时间（最后没数据，未获取到）。


SCHEDULER_DUPEFILTER_KEY = \'%(spider)s:dupefilter\'  # 去重规则，在redis中保存时对应的key


# 优先使用 DUPEFILTER_CLASS，如果没有就是用 SCHEDULER_DUPEFILTER_CLASS
SCHEDULER_DUPEFILTER_CLASS = \'scrapy_redis.dupefilter.RFPDupeFilter\'  # 去重规则对应处理的类

起始url

爬虫文件

import scrapy
from scrapy.http import Request
import scrapy_redis
from scrapy_redis.spiders import RedisSpider


class ChoutiSpider(RedisSpider): # 继承改为 用 RedisSpider
    name = \'chouti\'
    allowed_domains = [\'chouti.com\']

    # def start_requests(self): # 不再需要写 start_requests 了
    #     yield Request(url=\'https://dig.chouti.com\',callback=self.parse)


    def parse(self, response):
        print(response)

另写一个脚本文件 

import redis

conn = redis.Redis(host=\'127.0.0.1\',port=6379,password=\'yangtuo\')

# 通过往 redis 里面加数据从而控制爬虫的执行
# 如果没有数据就会夯住。一旦有数据就爬取，可以
conn.lpush(\'chouti:start_urls\',\'https://dig.chouti.com/r/pic/hot/1\')

代码流程

1.启动

scrapy crawl chouti --nolog

2.实例化调度器对象

找到  SCHEDULER = "scrapy_redis.scheduler.Scheduler"  配置并实例化调度器对象

　　- 执行Scheduler.from_crawler

　　- 执行Scheduler.from_settings
- 读取配置文件：

SCHEDULER_PERSIST    # 是否在关闭时候保留原来的调度器和去重记录，True=保留，False=清空
SCHEDULER_FLUSH_ON_START # 是否在开始之前清空 调度器和去重记录，True=清空，False=不清空
SCHEDULER_IDLE_BEFORE_CLOSE # 去调度器中获取数据时，如果为空，最多等待时间（最后没数据，未获取到）

- 读取配置文件：

SCHEDULER_QUEUE_KEY    # %(spider)s:requests    # 读取保存在 redis 中存放爬虫名字的 name 的值
SCHEDULER_QUEUE_CLASS    # scrapy_redis.queue.FifoQueue # 读取要使用的队列方式
SCHEDULER_DUPEFILTER_KEY # \'%(spider)s:dupefilter\'    # 读取保存在 redis 中存放爬虫去重规则名字的 key 的值
DUPEFILTER_CLASS    # \'scrapy_redis.dupefilter.RFPDupeFilter\'    # 读取去重的配置的类
SCHEDULER_SERIALIZER    # "scrapy_redis.picklecompat"    # 读取保存在 redis 的时候用的序列化方式

- 读取配置文件：

REDIS_HOST = \'140.143.227.206\' # 主机名
REDIS_PORT = 8888 # 端口
REDIS_PARAMS = {\'password\':\'beta\'} # Redis连接参数 默认：REDIS_PARAMS = {\'socket_timeout\': 30,\'socket_connect_timeout\': 30,\'retry_on_timeout\': True,\'encoding\': REDIS_ENCODING,}）
REDIS_ENCODING = "utf-8"

- 实例Scheduler对象

3. 爬虫开始执行起始URL

- 调用 scheduler.enqueue_requests()

def enqueue_request(self, request):
    # 请求是否需要过滤？ dont_filter = False 表示过滤 
    # 去重规则中是否已经有？（是否已经访问过，如果未访问添加到去重记录中。）
    # 要求过滤，且去重记录里面有就变表示访问过了
    if not request.dont_filter and self.df.request_seen(request):
        self.df.log(request, self.spider)
        return False     # 已经访问过就不要再访问了
    
    if self.stats:
        self.stats.inc_value(\'scheduler/enqueued/redis\', spider=self.spider)
    # print(\'未访问过，添加到调度器\', request)
    self.queue.push(request)
    return True

4. 下载器去调度器中获取任务，去下载

- 调用 scheduler.next_requests()

def next_request(self):
    block_pop_timeout = self.idle_before_close
    request = self.queue.pop(block_pop_timeout)
    if request and self.stats:
        self.stats.inc_value(\'scheduler/dequeued/redis\', spider=self.spider)
    return request

总结几个问题

1. 什么是深度优先？什么是广度优先？

深度优先 ：基于层级先进入到最深层级进行处理全部后再往上层级处理

广度优先 ：基于从第一层级开始，每次层级处理之后进入下一层及处理

2. scrapy中如何实现深度和广度优先？

先进先出，广度优先 FifoQueue

后进先出，深度优先 LifoQueue

优先级队列：

DEPTH_PRIORITY = 1 # 广度优先

DEPTH_PRIORITY = -1 # 深度优先

3. scrapy中 调度器 和 队列 和 dupefilter 的关系？

调度器，调配添加或获取那个request.

队列，存放request。

dupefilter，访问记录。

调度器拿到一个 request 要先去 dupefilter 里面看下有没有存在

　　如果存在就直接丢了

　　如果不存在才可以放在队列中

然后队列要基于自己的类型来进行相应规则的存放

打比方来说：

　　调度器 货车司机

　　dupefilter 仓库门卫

　　队列 仓库

　　司机要往仓库放货。先问门卫

　　　　门卫说里面有货了不让就没法放

　　　　门卫说可以才可以放

　　　　　　放货的时候仓库要按照自己的规则放货

　　司机要拉货 仓库就基于自己的规则直接给司机就行了

示例 - 腾讯招聘分布式爬虫

流程剖析

使用 scrapy-redis 配合 mongoDB 的分布式爬虫

代码

items.py

import scrapy


class TencentItem(scrapy.Item):
    # define the fields for your item here like:
    zh_name = scrapy.Field()
    zh_type = scrapy.Field()
    post_id = scrapy.Field()
    duty = scrapy.Field()
    require = scrapy.Field()

settings.py

BOT_NAME = \'Tencent\'

SPIDER_MODULES = [\'Tencent.spiders\']
NEWSPIDER_MODULE = \'Tencent.spiders\'

# Crawl responsibly by identifying yourself (and your website) on the user-agent
USER_AGENT = \'Mozilla/5.0\'

# Obey robots.txt rules
ROBOTSTXT_OBEY = False

# 使用scrapy_redis的调度器
SCHEDULER = "scrapy_redis.scheduler.Scheduler"

# 使用scrapy_redis的去重机制
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"

# 爬取完成后是否清除请求指纹
SCHEDULER_PERSIST = True
# \'scrapy_redis.pipelines.RedisPipeline\': 200
REDIS_HOST = \'172.40.91.129\'
REDIS_PORT = 6379


ITEM_PIPELINES = {
    \'Tencent.pipelines.TencentPipeline\': 300,
    # \'scrapy_redis.pipelines.RedisPipeline\': 200  # redis 的开启这个
    \'Tencent.pipelines.TencentMongoPipeline\': 200, 
}

pipline.py

import pymongo


class TencentPipeline(object):

def process_item(self, item, spider):
        print(dict(item))
        return item


class TencentMongoPipeline(object):
    def open_spider(self, spider):
        conn = pymongo.MongoClient(\'localhost\',27017)
        db = conn[\'tencent\']
        self.myset = db[\'job\']

    def process_item(self, item, spider):
        self.myset.insert_one(dict(item))

spider.py

# -*- coding: utf-8 -*-
import scrapy
import json
from ..items import TencentItem


class TencentSpider(scrapy.Spider):
    name = \'tencent\'
    allowed_domains = [\'tencent.com\']
    start_urls = [
        \'https://careers.tencent.com/tencentcareer/api/post/Query?timestamp=1557114143837&countryId=&cityId=&bgIds=&productId=&categoryId=&parentCategoryId=&attrId=&keyword=&pageIndex=1&pageSize=10&language=zh-cn&area=cn\']

    def parse(self, response):
        for page_index in range(1, 200):
            url = \'https://careers.tencent.com/tencentcareer/api/post/Query?timestamp=1557114143837&countryId=&cityId=&bgIds=&productId=&categoryId=&parentCategoryId=&attrId=&keyword=&pageIndex=%s&pageSize=10&language=zh-cn&area=cn\' % str(
                page_index)
            yield scrapy.Request(
                url=url,
                callback=self.parse_one_page
            )

    # 一级页面解析
    def parse_one_page(self, response):
        html = json.loads(response.text)

        for h in html[\'Data\'][\'Posts\']:
            item = TencentItem()

            item[\'zh_name\'] = h[\'RecruitPostName\']
            item[\'zh_type\'] = h[\'LocationName\']
            # 一级页面获取PostId,详情页URL需要此参数
            item[\'post_id\'] = h[\'PostId\']
            # 想办法获取到职位要求和职责,F12抓包,抓到地址
            two_url = \'https://careers.tencent.com/tencentcareer/api/post/ByPostId?timestamp=1557122746678&postId=%s&language=zh-cn\' % \\
                      item[\'post_id\']

            yield scrapy.Request(
                url=two_url,
                meta={\'item\': item},
                callback=self.parse_two_page
            )

    def parse_two_page(self, response):
        item = response.meta[\'item\']
        html = json.loads(response.text)
        # 职责
        item[\'duty\'] = html[\'Data\'][\'Responsibility\']
        # 要求
        item[\'require\'] = html[\'Data\'][\'Requirement\']

        yield item