| 爬虫教程（ 4 ） | 您所在的位置：网站首页 › scrapy性能提升 › 爬虫教程（ 4 ） |
爬虫教程（ 4 ）
|
scrapy 文档：https://docs.scrapy.org/en/latest/ scrapy-redis 文档：https://github.com/rmax/scrapy-redis/wiki/Usage 高性能爬虫 Scrapy 框架：https://www.cnblogs.com/wwg945/articles/9021888.html scrapy-redis 配置：https://www.cnblogs.com/wwg945/articles/9046232.html Scrapy爬虫入门教程：https://www.jianshu.com/p/43029ea38251 1、分布式 爬虫 scrapy 怎么 分布式 ？文档：http://doc.scrapy.org/en/master/topics/practices.html#distributed-crawls Scrapy 并没有提供内置的机制支持分布式(多服务器)爬取。不过还是有办法进行分布式爬取， 取决于您要怎么分布了。 如果有很多 spider，那分布负载最简单的办法就是启动多个Scrapyd，并分配到不同机器上。如果想要在多个机器上运行一个单独的spider，那您可以将要爬取的 url 进行分块，并发送给spider。...... scrapy-redis 分布式 爬虫框架scrapy 任务调度是基于文件系统，这样只能在单机执行 crawl。但是 scrapy-redis 巧妙的利用 redis 队列实现 request queue 和 items queue，利用 redis 的 set 实现 request 的去重，将 scrapy 从单台机器扩展多台机器，实现较大规模的爬虫集群 scrapy-redis 主要功能如下： 分布式爬虫。多个爬虫实例共享一个 redis request 队列，非常适合大范围多域名的爬虫集群分布式后处理。爬虫抓取到的 items push 到一个 redis items 队列，这就意味着可以开启多个 items processes 来处理抓取到的数据，比如存储到 Mongodb、Mysql基于 scrapy 即插即用组件。Scheduler + Duplication Filter、Item Pipeline、 Base Spiders scrapy 架构图 英文最新版文档：https://docs.scrapy.org/en/latest/topics/architecture.html中文 2.5 版本文档：https://www.osgeo.cn/scrapy/topics/architecture.html
Scrapy 中的数据流由执行引擎控制，流程如下： 引擎 首先从自己编写的 spider 中读取起始 url，然后封装成 Request对象引擎 把 "封装后的Request对象" 传递给 调度器 ( 调度器主要作用就是管理、调度url，可以简单的看作是一个 "间接的队列"，对 Requestd对象 管理、过滤 等操作)。引擎 请求 调度器，调度器返回一个 Request对象 给引擎。引擎 将 Request对象 发送到下载器。下载器会将请求通过下载器中间件。( process_request() )下载器完成页面下载后，下载器会将生成的 响应Response 通过下载器中间件（process_response（）），最后将其发送到引擎。引擎接收来自下载器的 响应 并将其发送给 自己编写的 spider 进行处理，但是在发送之前会先 传递 通过spider中间件（参见process_spider_input（））。自己编写的 spider 处理响应并返回 "抓取的数据Item" 及(跟进的)新的Request给引擎，通过蜘蛛中间件（参见process_spider_output（））。引擎将 "抓取的数据Item" 发送到 pipeline，将 "新的请求" 发送到 调度程序。并继续从调度器中获取 下一个 "Request对象" 来抓取。该过程重复（从步骤 3 开始），直到没有更多地 request对象 ，最后关闭引擎。整个工作流程 1.引擎 将爬虫中起始的url构造成request对象，并传递给调度器。2.引擎 从 调度器 中获取到request对象然后交给下载器。3.由 下载器 来获取到页面源代码，并封装成response对象，并返回给引擎。4.引擎 将获取到的response对象传递给 spider，由 spider 对数据进行解析(parse)，并返回给引擎5.引擎将数据传递给 pipeline 进行数据持久化保存或进一步的数据处理6.在此期间如果spider中提取到的并不是数据。而是子页面ur.可以进一步提交给调度器，进而重复 步骤2 的过程 scrapy-redis 安装文档：https://scrapy-redis.readthedocs.org. 安装 scrapy-redis：pip install scrapy-redis scrapy-redis 源码 分析
可以看到 scrapy-redis 的 spiders.py 模块，导入了 scrapy.spiders 的 Spider、CrawlSpider，然后重新写了两个类 RedisSpiders、RedisCrawlSpider，分别继承 Spider、CrawlSpider，所以如果要想从 redis 读取任务，需要把自己写的 spider 继承 RedisSpiders、RedisCrawlSpider，而不是 scrapy 的 Spider、CrawlSpider。。。 spider.py 中 RedisMixin、RedisSpiders、RedisCrawlSpider RedisMixin 类，读取 配置文件，决定使用 什么类型的 redis 队列
可以在 setting 中设置下面两项，来决定 redis 任务队列是 set 还是 zset REDIS_START_URLS_AS_SET REDIS_START_URLS_AS_ZSET Scrapy-redis 之 RFPDupeFilter、Queue、Scheduler：https://www.cnblogs.com/Alexephor/p/11446167.html 1.找到from scrapy_redis.scheduler import Scheduler -执行Scheduler.from_crawler -执行Scheduler.from_settings - 读取配置文件： SCHEDULER_PERSIST # 是否在关闭时候保留原来的调度器和去重记录，True=保留，False=清空 SCHEDULER_FLUSH_ON_START # 是否在开始之前清空 调度器和去重记录，True=清空，False=不清空 SCHEDULER_IDLE_BEFORE_CLOSE # 去调度器中获取数据时，如果为空，最多等待时间（最后没数据，未获取到） - 读取配置文件： SCHEDULER_QUEUE_KEY # 调度器中请求存放在redis中的key SCHEDULER_QUEUE_CLASS # 这里可以选择三种先进先出、后进先出、优先级，默认使用优先级队列（默认），其他：PriorityQueue（有序集合），FifoQueue（列表）、LifoQueue（列表） SCHEDULER_DUPEFILTER_KEY # 去重规则，在redis中保存时对应的key DUPEFILTER_CLASS # 这里有两种选择使用默认或者自己定义的 # 内置比如：DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter' # 自定义的比如：DUPEFILTER_CLASS = 'redisdepth.xxx.DupeFilter' 这个优先级别高 在源码里边是先判断然后再后续操作 SCHEDULER_SERIALIZER # 对保存到redis中的数据进行序列化，默认使用pickle - 读取配置文件：redis-server # 源码在connection.py中17行 REDIS_HOST = '192.168.1.13' # 主机名 REDIS_PORT = 3306 # 端口 REDIS_PARAMS = {'password': 'woshinidaye'} # Redis连接参数 默认：REDIS_PARAMS = {'socket_timeout': 30,'socket_connect_timeout': 30,'retry_on_timeout': True,'encoding': REDIS_ENCODING,}） # REDIS_PARAMS['redis_cls'] = 'myproject.RedisClient' # 指定连接Redis的Python模块 默认：redis.StrictRedis REDIS_ENCODING = "utf-8" # redis编码类型 默认：'utf-8' # REDIS_URL = 'redis://user:pass@hostname:9001' # 连接URL（优先于以上配置）源码可以看到 2.爬虫开始执行起始URL - 调用Scheduler.enqueue_request def enqueue_request(self, request): # 请求需要过滤？并且 去重规则是否已经有？（是否已经访问，如果未访问添加到去重记录）request_seen去重规则重要的一个方法 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) # 未访问过，添加到调度器中把这个请求 self.queue.push(request) return True 3.下载器去调度中获取任务，去执行任务下载 - 调用Scheduler.next_request 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 requestsettings需要的配置 # redis去重配置 REDIS_HOST = '192.168.1.13' # 主机名 REDIS_PORT = 3306 # 端口 REDIS_PARAMS = {'password': 'woshinidaye'} # Redis连接参数 默认：REDIS_PARAMS = {'socket_timeout': 30,'socket_connect_timeout': 30,'retry_on_timeout': True,'encoding': REDIS_ENCODING,}） # REDIS_PARAMS['redis_cls'] = 'myproject.RedisClient' # 指定连接Redis的Python模块 默认：redis.StrictRedis REDIS_ENCODING = "utf-8" # redis编码类型 默认：'utf-8' # REDIS_URL = 'redis://user:pass@hostname:9001' # 连接URL（优先于以上配置）源码可以看到 DUPEFILTER_KEY = 'dupefilter:%(timestamp)s' # 纯源生的它内部默认是用的以时间戳作为key # DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter' # 我自定义在源码之上改了保存在redis中的key配置 DUPEFILTER_CLASS = 'redisdepth.xxx.RedisDupeFilter' # 自定义redis去重配置 # DUPEFILTER_CLASS = 'redisdepth.xxx.DupeFilter' # #############调度器配置########################### # from scrapy_redis.scheduler import Scheduler SCHEDULER = "scrapy_redis.scheduler.Scheduler" DEPTH_PRIORITY = 1 # 广度优先 # DEPTH_PRIORITY = -1 # 深度优先 SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue' # 默认使用优先级队列（默认），其他：PriorityQueue（有序集合），FifoQueue（列表）、LifoQueue（列表） # 广度优先 # SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.FifoQueue' # 深度优先 # SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.LifoQueue' SCHEDULER_QUEUE_KEY = '%(spider)s:requests' # 调度器中请求存放在redis中的key SCHEDULER_SERIALIZER = "scrapy_redis.picklecompat" # 对保存到redis中的数据进行序列化，默认使用pickle SCHEDULER_PERSIST = True # 是否在关闭时候保留原来的调度器和去重记录，True=保留，False=清空 SCHEDULER_FLUSH_ON_START = True # 是否在开始之前清空 调度器和去重记录，True=清空，False=不清空 SCHEDULER_IDLE_BEFORE_CLOSE = 10 # 去调度器中获取数据时，如果为空，最多等待时间（最后没数据，未获取到）。 SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter' # 去重规则，在redis中保存时对应的key SCHEDULER_DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter' # 去重规则对应处理的类配置文件大解读 # -*- coding: utf-8 -*- # 爬虫名称 BOT_NAME = 'redisdepth' # 爬虫应用路径 SPIDER_MODULES = ['redisdepth.spiders'] NEWSPIDER_MODULE = 'redisdepth.spiders' # Crawl responsibly by identifying yourself (and your website) on the user-agent # 客服端user-agent请求头 #USER_AGENT = 'redisdepth (+http://www.yourdomain.com)' USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36' # 爬虫君子证书，禁止爬虫设置 # Obey robots.txt rules # ROBOTSTXT_OBEY = True ROBOTSTXT_OBEY = False # Configure maximum concurrent requests performed by Scrapy (default: 16) # 并发请求数 力度要粗点 #CONCURRENT_REQUESTS = 32 # Configure a delay for requests for the same website (default: 0) # See https://docs.scrapy.org/en/latest/topics/settings.html#download-delay # See also autothrottle settings and docs # 延迟下载秒数 #DOWNLOAD_DELAY = 3 # The download delay setting will honor only one of: # 单域名访问并发数 并且延迟下次秒数也用在每个域名 #CONCURRENT_REQUESTS_PER_DOMAIN = 16 # 单IP访问并发数，如果有值则忽略：CONCURRENT_REQUESTS_PER_DOMAIN，并且延迟下次秒数也应用在每个IP #CONCURRENT_REQUESTS_PER_IP = 16 # Disable cookies (enabled by default) # 是否支持cookie，cookiejar进行操作cookie #COOKIES_ENABLED = False # Disable Telnet Console (enabled by default) # Telnet用于查看当前爬虫的信息，操作爬虫等... # 使用telnet ip port ，然后通过命令操作 # TELNETCONSOLE_ENABLED = True # TELNETCONSOLE_HOST = '127.0.0.1' # TELNETCONSOLE_PORT = [6023,] #TELNETCONSOLE_ENABLED = False # 默认请求头 # Override the default request headers: #DEFAULT_REQUEST_HEADERS = { # 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', # 'Accept-Language': 'en', #} # 爬虫中间件 # Enable or disable spider middlewares # See https://docs.scrapy.org/en/latest/topics/spider-middleware.html # SPIDER_MIDDLEWARES = { # # 'redisdepth.middlewares.RedisdepthSpiderMiddleware': 543, # 'redisdepth.sd.Sd1': 666, # 'redisdepth.sd.Sd2': 667, # # } # 下载中间件 # Enable or disable downloader middlewares # See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html # DOWNLOADER_MIDDLEWARES = { # # 'redisdepth.middlewares.RedisdepthDownloaderMiddleware': 543, # # 'redisdepth.md.Md1': 666, # # 'redisdepth.md.Md2': 667 # } # 自定义扩展，基于信号进行调用 # Enable or disable extensions # See https://docs.scrapy.org/en/latest/topics/extensions.html EXTENSIONS = { # 'scrapy.extensions.telnet.TelnetConsole': None, 'redisdepth.ext.MyExtension': 666, } # 定义pipeline处理请求 # Configure item pipelines # See https://docs.scrapy.org/en/latest/topics/item-pipeline.html #ITEM_PIPELINES = { # 'redisdepth.pipelines.RedisdepthPipeline': 300, #} """ 自动限速算法 from scrapy.contrib.throttle import AutoThrottle 自动限速设置 1. 获取最小延迟 DOWNLOAD_DELAY 2. 获取最大延迟 AUTOTHROTTLE_MAX_DELAY 3. 设置初始下载延迟 AUTOTHROTTLE_START_DELAY 4. 当请求下载完成后，获取其"连接"时间 latency，即：请求连接到接受到响应头之间的时间 5. 用于计算的... AUTOTHROTTLE_TARGET_CONCURRENCY target_delay = latency / self.target_concurrency new_delay = (slot.delay + target_delay) / 2.0 # 表示上一次的延迟时间 new_delay = max(target_delay, new_delay) new_delay = min(max(self.mindelay, new_delay), self.maxdelay) slot.delay = new_delay """ # Enable and configure the AutoThrottle extension (disabled by default) # See https://docs.scrapy.org/en/latest/topics/autothrottle.html # 开始自动限速 #AUTOTHROTTLE_ENABLED = True # The initial download delay # 初始下载延迟 #AUTOTHROTTLE_START_DELAY = 5 # The maximum download delay to be set in case of high latencies # 最大下载延迟 #AUTOTHROTTLE_MAX_DELAY = 60 # The average number of requests Scrapy should be sending in parallel to # each remote server # 平均每秒并发数 #AUTOTHROTTLE_TARGET_CONCURRENCY = 1.0 # Enable showing throttling stats for every response received: # 是否显示 #AUTOTHROTTLE_DEBUG = False # Enable and configure HTTP caching (disabled by default) # See https://docs.scrapy.org/en/latest/topics/downloader-middleware.html#httpcache-middleware-settings """ 启用缓存 目的用于将已经发送的请求或相应缓存下来，以便以后使用 from scrapy.downloadermiddlewares.httpcache import HttpCacheMiddleware from scrapy.extensions.httpcache import DummyPolicy from scrapy.extensions.httpcache import FilesystemCacheStorage """ # 是否启用缓存策略 #HTTPCACHE_ENABLED = True # 缓存策略：所有请求均缓存，下次在请求直接访问原来的缓存即可 # HTTPCACHE_POLICY = "scrapy.extensions.httpcache.DummyPolicy" # 缓存策略：根据Http响应头：Cache-Control、Last-Modified 等进行缓存的策略 # HTTPCACHE_POLICY = "scrapy.extensions.httpcache.RFC2616Policy" # 缓存超时时间 #HTTPCACHE_EXPIRATION_SECS = 0 # 缓存保存路径 #HTTPCACHE_DIR = 'httpcache' # 缓存忽略的http状态码 #HTTPCACHE_IGNORE_HTTP_CODES = [] # 缓存存储的插件 #HTTPCACHE_STORAGE = 'scrapy.extensions.httpcache.FilesystemCacheStorage' scrapy-redis 调度器(Scheduler) scrapy-redis 调度器 通过 redis 的 set 不重复的特性，巧妙的实现了 Duplication Filter 去重（DupeFilter set 存放爬取过的 request）。Spider 新生成的 request，将 request 的指纹到 redis 的 DupeFilter set 检查是否重复，并将不重复的 request push 写入 redis 的 request 队列。调度器每次从 redis 的 request 队列里根据优先级 pop 出一个 request, 将此 request 发给 spider 处理。
将 Spider 爬取到的 Item 给 scrapy-redis 的 Item Pipeline，将爬取到的 Item 存入 redis 的 items 队列。可以很方便的从 items 队列中提取 item，从而实现 items processes 集群
：https://www.51tietu.net/xiaohua/ scrapy_redis.spiders 下有两个类 RedisSpider 和 RedisCrawlSpider，能够使 spider 从 Redis 读取 start_urls，然后执行爬取，若爬取过程中返回更多的 request url，那么它会继续进行直至所有的 request 完成之后，再从 redis start_urls 中读取下一个 url，循环这个过程 创建 scrapy-redis 项目 方法 1：命令行执行：scrapy startproject MyScrapyRedis，然后自己写的 spider 继承 RedisSpider 或者 RedisCrawlSpider ，设置对应的 redis_key ，即队列的在 redis 中的 key。注意：这个需要手动 在 setting.py 里面配置设置。（ 参考配置：https://github.com/rmax/scrapy-redis ） 方法 2：使用 scrapy-redis 的 example 来修改。先从 github ( https://github.com/rmax/scrapy-redis ) 上拿到 scrapy-redis 的 example，然后将里面的 example-project 目录移到指定的地址。
使用 PyCharm 打开 MyScrapyRedis 项目，继承 scrapy_redis 的 RedisSpider 类 scrapy_redis 的 RedisSpider 类 说明：
可以看到 RedisSpider 有三个属性，这三个属性的默认值都在 scrapy_redis 下的 default.py 中 redis_key redis_batch_sizeredis_encodingsettings 需要设置的选项 REDIS_START_URLS_KEY ( 如果设置了redis_key ，则覆盖这个配置 )REDIS_START_URLS_BATCH_SIZE ( 已经废弃，使用 CONCURRENT_REQUESTS 代替 )REDIS_START_URLS_AS_SET ( 如果是 True 则使用 set 集合作为 任务队列，默认 False 使用 list )REDIS_ENCODING ( 设置 队列任务的编码 ) 编写 example.py 爬虫文件这里设置 redis_key = f'redis_key:{name}'，同时设置 REDIS_START_URLS_AS_SET= True import scrapy from scrapy_redis.spiders import RedisSpider, RedisCrawlSpider class ExampleSpider(RedisSpider): name = "example" redis_key = f'redis_key:{name}_zset' custom_settings = { 'REDIS_START_URLS_AS_ZSET': True, # 'REDIS_START_URLS_AS_SET': True, # 'SCHEDULER_IDLE_BEFORE_CLOSE': 5 } def parse(self, response): print(f'response.url ---> {response.url}') item = { 'url': response.url, } yield item pass if __name__ == '__main__': # add_task() from scrapy import cmdline cmdline.execute('scrapy crawl example'.split()) pass 编写 ItemScrapy 中可以直接返回一个 Python 的 字典 给 pipeline，但是这并不是最佳实践。scrapy提供了一个Item基类，可以通过继承这个类定义自己的结构化数据，比到处传递字典更好。 下面是官方文档的例子。 import scrapy class Product(scrapy.Item): name = scrapy.Field() price = scrapy.Field() stock = scrapy.Field() last_updated = scrapy.Field(serializer=str)一般都定义在 scrapy 项目的 items.py 文件中。定义好之后，在爬虫中我们就不应该在返回字典了，而是初始化并返回我们自定义的 Item 对象。 提示：为了演示，下面示例代码是直接返回 Python 字典给 pipeline 修改 settings.pyscrapy-redis 的默认配置 ：https://github.com/rmax/scrapy-redis/blob/master/src/scrapy_redis/defaults.py import redis # For standalone use. DUPEFILTER_KEY = 'dupefilter:%(timestamp)s' PIPELINE_KEY = '%(spider)s:items' STATS_KEY = '%(spider)s:stats' REDIS_CLS = redis.StrictRedis REDIS_ENCODING = 'utf-8' # Sane connection defaults. REDIS_PARAMS = { 'socket_timeout': 30, 'socket_connect_timeout': 30, 'retry_on_timeout': True, 'encoding': REDIS_ENCODING, } REDIS_CONCURRENT_REQUESTS = 16 SCHEDULER_QUEUE_KEY = '%(spider)s:requests' SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.PriorityQueue' SCHEDULER_DUPEFILTER_KEY = '%(spider)s:dupefilter' SCHEDULER_DUPEFILTER_CLASS = 'scrapy_redis.dupefilter.RFPDupeFilter' SCHEDULER_PERSIST = False START_URLS_KEY = '%(name)s:start_urls' START_URLS_AS_SET = False START_URLS_AS_ZSET = False MAX_IDLE_TIME = 0 修改项目 MyScrapyRedis 目录下 setting.py 文件。下面列举了修改后的配置文件中与 scrapy-redis 有关的部分，middleware、proxy 等在此略过。 # 指定使用 scrapy-redis 的 Scheduler SCHEDULER = "scrapy_redis.scheduler.Scheduler" # 指定使用 scrapy-redis 的 RFPDupeFilter DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter" # 在 redis 中保持 scrapy-redis 用到的各个队列，从而允许暂停和暂停后恢复 SCHEDULER_PERSIST = True # 指定排序爬取地址时使用的队列，默认是按照优先级排序 SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderPriorityQueue" # SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderQueue" # SCHEDULER_QUEUE_CLASS = "scrapy_redis.queue.SpiderStack" # 只在使用 SpiderQueue 或者 SpiderStack 是有效的参数,指定爬虫关闭的最大空闲时间 SCHEDULER_IDLE_BEFORE_CLOSE = 10 ITEM_PIPELINES = { 'MyScrapyRedis.pipelines.ExamplePipeline': 300, 'MyScrapyRedis.pipelines.MyRedisPipeline': 400, # 'scrapy_redis.pipelines.RedisPipeline': 400, } LOG_LEVEL = 'DEBUG' # 指定redis的连接参数 REDIS_HOST = '127.0.0.1' REDIS_PORT = 6379 REDIS_PARAMS = {} # REDIS_URL = 'redis://user:pass@hostname:9001' # 连接URL (优先于以上配置) 源码可以看到 新建 pipeline 继承 scrapy_redis 的 pipeline # Define your item pipelines here # # Don't forget to add your pipeline to the ITEM_PIPELINES setting # See: http://doc.scrapy.org/topics/item-pipeline.html import json from datetime import datetime from scrapy_redis.pipelines import RedisPipeline class ExamplePipeline(object): def process_item(self, item, spider): item["crawled"] = str(datetime.now().replace(microsecond=0)) item["spider"] = spider.name return item class MyRedisPipeline(RedisPipeline): def _process_item(self, item, spider): key = self.item_key(item, spider) # data = self.serialize(item) self.server.rpush(key, json.dumps(item, ensure_ascii=False)) return item也可以不用重写，通过在 setting.py 里面配置 REDIS_ITEMS_SERIALIZER = 'json.dumps' 即可使用 json 序列化（ 通过查看 scrapy-redis 的 pipelines.py 可以知道：scrapy-redis 默认使用 ScrapyJSONEncoder 进行项目序列化 ） 参考：https://www.cnblogs.com/Alexephor/p/11446167.html 将数据保存到 MongoDB 的管道Item Pipeline：https://docs.scrapy.org/en/latest/topics/item-pipeline.html
管道除了验证数据，还可以将数据保存到数据库中。这时候仅仅一个process_item(self, item, spider)函数就不够了。所以操作数据库的管道还应该包含几个函数用于建立和关闭数据库连接。 下面的例子也是 scrapy 官方文档的例子，演示了持久化数据管道的用法。这个管道是从类方法from_crawler(cls, crawler)中初始化出来的，该方法实际上读取了scrapy的配置文件。这和直接将数据库连接写在代码中相比，是更加通用的方式。初始化和关闭数据库连接的操作都在对应的方法中执行。 import pymongo class MongoPipeline(object): collection_name = 'scrapy_items' def __init__(self, mongo_uri, mongo_db): self.mongo_uri = mongo_uri self.mongo_db = mongo_db @classmethod def from_crawler(cls, crawler): return cls( mongo_uri=crawler.settings.get('MONGO_URI'), mongo_db=crawler.settings.get('MONGO_DATABASE', 'items') ) def open_spider(self, spider): self.client = pymongo.MongoClient(self.mongo_uri) self.db = self.client[self.mongo_db] def close_spider(self, spider): self.client.close() def process_item(self, item, spider): self.db[self.collection_name].insert_one(dict(item)) return item 使用 "文件、图片" 管道
除了自己编写管道之外，scrapy 还预定义了几个管道，可以帮助我们方便的保存文件和图片。这些管道有以下特点： 可以避免重复下载最近的文件。指定文件保存位置（文件系统或者亚马逊S3）对于图片管道来说还有额外功能： 将图片转换成常见格式（JPG）和模式（RGB）生成图片缩略图只下载大于某长宽的图片使用文件管道的过程如下： 首先需要Item类中有file_urls和files两个属性，然后在爬虫中将想爬取的文件地址放到file_urls属性中，然后返回在Item传递到文件管道的时候，调度程序会用下载器将地址对应的文件下载下来，将文件属性（包括保存路径等）放到files属性中，file_urls和files中是一一对应的使用图片管道的过程是相似的，不过要操作的属性是image_urls和images。 如果你不想使用这几个属性，其实属性名也是可以修改的，需要修改下面四个属性。 FILES_URLS_FIELD = 'field_name_for_your_files_urls' FILES_RESULT_FIELD = 'field_name_for_your_processed_files' IMAGES_URLS_FIELD = 'field_name_for_your_images_urls' IMAGES_RESULT_FIELD = 'field_name_for_your_processed_images'要启用文件管道和图片管道，同样需要激活，当然如果同时激活这两个管道也是可行的。 ITEM_PIPELINES = {'scrapy.pipelines.images.ImagesPipeline': 1} # 或者 ITEM_PIPELINES = {'scrapy.pipelines.files.FilesPipeline': 1}文件和图片保存位置需要分别指定。 FILES_STORE = '/path/to/valid/dir' IMAGES_STORE = '/path/to/valid/dir'文件和图片管道可以避免下载最近的文件，对应的文件过期时间也可以配置，单位是天。 # 120 days of delay for files expiration FILES_EXPIRES = 120 # 30 days of delay for images expiration IMAGES_EXPIRES = 30图片管道可以在保存图片的时候同时生成缩略图，缩略图配置是一个字典，键是缩略图的名字，值是缩略图长和宽。 IMAGES_THUMBS = { 'small': (50, 50), 'big': (270, 270), }最后图片会保存成下面这样，图片的文件名是图片路径的SHA1哈希值。 /图片保存路径/full/完整图片.jpg /图片保存路径/thumbs/small/小图片.jpg /图片保存路径/thumbs/big/中图片.jpg如果不想使用SHA1哈希值作为文件名，可以继承ImagesPipeline基类并重写file_path函数，这里是另外一位简书作者的爬虫项目，他重写了这个函数。我们可以作为参考。 如果要过滤小图片，启用下面的配置。默认情况下对图片尺寸没有约束，所以所有图片都会下载。 IMAGES_MIN_HEIGHT = 110 IMAGES_MIN_WIDTH = 110默认情况下文件和图片管道不支持重定向，遇到需要重定向的链接意味着下载失败，不过我们也可以启用重定向。 MEDIA_ALLOW_REDIRECTS = True 添加任务添加任务到 redis 的 list 中： 通过命令添加 urls 到 redis：redis-cli lpush myspider:start_urls https://baidu.com 通过代码 添加任务到 redis 的 list 中 import json from scrapy.utils.project import get_project_settings from scrapy_redis.connection import get_redis_from_settings from scrapy_redis import connection from scrapy_redis.queue import PriorityQueue # def _encode_request(self, request): # """Encode a request object""" # obj = request_to_dict(request, self.spider) # return self.serializer.dumps(obj) # # # def _decode_request(self, encoded_request): # """Decode an request previously encoded""" # obj = self.serializer.loads(encoded_request) # return request_from_dict(obj, self.spider) def add_task_to_redis(): redis_key = 'start_urls:yy_spider_request' url_string = 'http://www.youyuan.com/find/beijing/mm18-25/advance-0-0-0-0-0-0-0/p1/' # 方法 1 server = get_redis_from_settings(get_project_settings()) server.lpush(redis_key, url_string) # server.zadd(redis_key, url_string, 1000) # 方法 2 # server = connection.from_settings(get_project_settings()) # server.execute_command('ZADD', redis_key, 1000, url_string) if __name__ == '__main__': # temp = 'test json string' # print(json.dumps(temp)) add_task_to_redis() pass添加 "json 格式的任务" 到 redis 的 set、zset 中 import json import time from datetime import datetime from scrapy_redis.connection import get_redis name = "example" redis_key_set = f'redis_key:{name}' redis_key_zset = f'redis_key:{name}_zset' redis_config = { 'host': '127.0.0.1', 'port': 6379, 'db': 0 } redis_conn = get_redis(**redis_config) url_list = [ 'https://www.51tietu.net/xiaohua/' ] def add_task_to_set(): for index in range(1, 10): url = f'https://www.51tietu.net/xiaohua/{index}' url_list.append(url) for url in url_list: redis_conn.sadd(redis_key_set, json.dumps({'url': url}, ensure_ascii=False)) print(f'add url ---> {url}') def add_task_to_zset(): for index in range(1, 10): url = f'https://www.51tietu.net/xiaohua/{index}' url_list.append(url) for url in url_list: redis_conn.execute_command( 'ZADD', redis_key_zset, # 用 15 位时间戳作为 score int(datetime.now().timestamp() * 100000), json.dumps({'url': url}, ensure_ascii=False) ) print(f'add url ---> {url}') if __name__ == '__main__': add_task_to_set() add_task_to_zset() pass
这里直接执行上面的 example.py。运行结果：
redis 中保存的 结果如下：
启用 DownLoader 中间件 DOWNLOADER_MIDDLEWARES = { 'MyScrapyRedis.middlewares.ProxyMiddleware': 400, } scrapy内置了14个下载器中间件，
详情参考文档。如果希望禁用某些内置的中间件，可以将值设置为 None。 编写自己的下载器中间件 自定义下载器中间件应该继承 scrapy.downloadermiddlewares.DownloaderMiddleware 类，该类有如下几个方法，用于操纵请求和响应，我们只要重写这几个方法即可。这几个方法的作用请参考 官方文档 ( https://doc.scrapy.org/en/latest/topics/downloader-middleware.html )，它们比较复杂，所以我就不说了。 process_request(request, spider)process_response(request, response, spider)process_exception(request, exception, spider)创建 middlewares.py 并编辑 (settings.py 同级目录)
有这么一个场景，有些请求是不需要代理IP，怎么才能让它请求超时的时候，再使用代理池的IP地址进行重新请求呢？ 1、scrapy的基本请求步骤是，首先执行父类里面（scrapy.Spider）里面的start_requests方法，2、然后start_requests方法也是取拿我们设置的start_urls变量里面的url地址3、最后才执行make_requests_from_url方法，并只传入一个url变量那么，我们就可以重写make_requests_from_url方法，从而直接调用scrapy.Request()方法。参数说明 1、url=url,其实就是最后start_requests()方法里面拿到的url地址2、meta这里我们只设置了一个参数，download_timeout:10，作用就是当第一次发起请求的时候，等待10秒钟，如果没有请求成功的话，就会直接执行download_middleware里面的方法，我们下面介绍。3、callback回调函数，其实就是本次的本次所有操作完成后执行的操作，注意，这里可不是说执行完上面所有操作后，再执行这个操作，比如说请求了一个url，并且成功了，下面就会执行这个方法。4、dont_filter=False，这个很重要，有人说过不加的话默认就是False，但是亲测必须得加，作用就是scrapy默认有去重的方法，等于False的话就意味着不参加scrapy的去重操作。亲测，请求一个页面，拿到第一个页面后，抓取想要的操作后，第二页就不行了，只有加上它才可以。 import scrapy class HttpbinTestSpider(scrapy.Spider): name = "httpbin_test" allowed_domains = ["httpbin.ort/get"] start_urls = ['http://httpbin.org/get'] def make_requests_from_url(self,url): self.logger.debug('Try first time') return scrapy.Request( url=url, meta={'download_timeout':10}, callback=self.parse, dont_filter=False ) def parse(self, response): print(response.text) class HttpbinProxyMiddleware(object): logger = logging.getLogger(__name__) # def process_request(self, request, spider): # # pro_addr = requests.get('http://127.0.0.1:5000/get').text # # requesta['proxy'] = 'http://' + pro_addr # pass # # def process_response(self, request, response, spider): # # 可以拿到下载完的response内容，然后对下载完的内容进行修改等操作。 # pass def process_exception(self, request, response, spider): self.logger.debug('Try Exception time') self.logger.debug('Try second time') proxy_addr = requests.get('http://127.0.0.1:5000/get').text self.logger.debug(proxy_addr) requesta['proxy'] = 'http://{0}'.format(proxy_addr) 在scrapy中的中间件里面，对应的中间件后面的数字越小，执行优先级越高。 DOWNLOADER_MIDDLEWARES = { 'httpbin.middlewares.HttpbinProxyMiddleware': 543, #设置不参与scrapy的自动重试的动作 'scrapy.downloadermiddlewares.retry.RetryMiddleware':None }实际爬虫过程中如果请求过于频繁，通常会被临时重定向到登录页面即302，甚至是提示禁止访问即403，因此可以对这些响应执行一次代理请求： (1) 参考原生redirect.py 模块，满足dont_redirect 或handle_httpstatus_list 等条件时，直接传递response (2) 不满足条件(1)，如果响应状态码为 302 或 403，使用代理重新发起请求 (3) 使用代理后，如果响应状态码仍为 302 或 403，直接丢弃 from w3lib.url import safe_url_string from six.moves.urllib.parse import urljoin from scrapy.exceptions import IgnoreRequest class MyAutoProxyDownloaderMiddleware(object): def __init__(self, settings): self.proxy_status = settings.get('PROXY_STATUS', [302, 403]) self.proxy_config = settings.get('PROXY_CONFIG', 'http://username:password@some_proxy_server:port') @classmethod def from_crawler(cls, crawler): return cls(settings=crawler.settings) def process_response(self, request, response, spider): req_meta = requesta if (req_meta.get('dont_redirect', False) or response.status in getattr(spider, 'handle_httpstatus_list', []) or response.status in req_meta.get('handle_httpstatus_list', []) or req_meta.get('handle_httpstatus_all', False)): return response if response.status in self.proxy_status: if 'Location' in response.headers: location = safe_url_string(response.headers['location']) redirected_url = urljoin(request.url, location) else: redirected_url = '' # AutoProxy for first time if not requesta.get('auto_proxy'): requesta.update({'auto_proxy': True, 'proxy': self.proxy_config}) new_request = request.replace(meta=requesta, dont_filter=True) new_request.priority = request.priority + 2 spider.log(f'Will AutoProxy for {redirected_url}') return new_request # IgnoreRequest for second time else: spider.logger.warn( '忽略 response : 代理后status_code 仍在{self.proxy_status}') raise IgnoreRequest return response项目 settings.py 添加代码，注意必须在默认的 RedirectMiddleware 和 HttpProxyMiddleware 之间。 DOWNLOADER_MIDDLEWARES = { # 'scrapy.downloadermiddlewares.redirect.RedirectMiddleware': 600, 'my_middlewares.MyAutoProxyDownloaderMiddleware': 601, # 'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 750, } PROXY_STATUS = [302, 403] PROXY_CONFIG = 'http://username:password@some_proxy_server:port' 示例： class HttpProxymiddleware(object): # 一些异常情况汇总 EXCEPTIONS_TO_CHANGE = ( defer.TimeoutError, TimeoutError, ConnectionRefusedError, ConnectError, ConnectionLost, TCPTimedOutError, ConnectionDone) def __init__(self): self.redis = redis.from_url( 'redis://:你的密码@localhost:6379/0', decode_responses=True ) pass def process_request(self, request, spider): #拿出全部 key，随机选取一个键值对 keys = self.rds.hkeys("xila_hash") key = random.choice(keys) #用eval函数转换为dict proxy = eval(self.rds.hget("xila_hash",key)) logger.warning(f"{str(proxy)}使用中") #将代理 ip 和 key 存入 mate requesta["proxy"] = proxy["ip"] requesta["accountText"] = key def process_response(self, request, response, spider): http_status = response.status #根据response的状态判断 ，200的话ip的times +1重新写入数据库，返回response到下一环节 if http_status == 200: key = requesta["accountText"] proxy = eval(self.rds.hget("xila_hash",key)) proxy["times"] = proxy["times"] + 1 self.rds.hset("xila_hash",key,proxy) return response #403有可能是因为user-agent不可用引起，和代理ip无关，返回请求即可 elif http_status == 403: logging.warning("403重新请求中") return request.replace(dont_filter=True) #其他情况姑且被判定ip不可用，times小于10的，删掉，大于等于10的暂时保留 else: ip = requesta["proxy"] key = requesta["accountText"] proxy = eval(self.rds.hget("xila_hash", key)) if proxy["times"] < 10: self.rds.hdel("xila_hash",key) logging.warning(f"{ip}不可用") return request.replace(dont_filter=True) def process_exception(self, request, exception, spider): #其他一些timeout之类异常判断后的处理，ip不可用删除即可 if isinstance(exception, self.EXCEPTIONS_TO_CHANGE) \ and requesta.get('proxy', False): key = requesta["accountText"] print("{key}不可用, 将被删除") proxy = eval(self.rds.hget("xila_hash", key)) if proxy["times"] < 10: self.rds.hdel("xila_hash", key) logger.debug(f"Proxy {requesta['proxy']}链接出错{exception}") return request.replace(dont_filter=True) 伪造 User-Agentfake-useragent 下载 和 使用方法：https://pypi.org/project/fake-useragent 方法 1： 在setting.py文件中加入以下内容，这是一些浏览器的头信息 USER_AGENT_LIST = [ "Avant Browser/1.2.789rel1 (http://www.avantbrowser.com)", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/534.27 (KHTML, like Gecko) Chrome/12.0.712.0 Safari/534.27", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/535.1 (KHTML, like Gecko) Chrome/13.0.782.24 Safari/535.1", "Mozilla/5.0 (Windows NT 6.0) AppleWebKit/535.2 (KHTML, like Gecko) Chrome/15.0.874.120 Safari/535.2", "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/535.7 (KHTML, like Gecko) Chrome/16.0.912.36 Safari/535.7", "Mozilla/5.0 (Windows; U; Windows NT 6.0 x64; en-US; rv:1.9pre) Gecko/2008072421 Minefield/3.0.2pre", "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.0.10) Gecko/2009042316 Firefox/3.0.10", "Mozilla/5.0 (Windows; U; Windows NT 6.0; en-US; rv:1.9.1.6) Gecko/20091201 Firefox/3.5.6 GTB5", "Mozilla/5.0 (Windows NT 6.1; rv:2.0.1) Gecko/20100101 Firefox/4.0.1", "Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1", "Mozilla/5.0 (Windows NT 5.1; rv:5.0) Gecko/20100101 Firefox/5.0", "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:6.0a2) Gecko/20110622 Firefox/6.0a2", "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:7.0.1) Gecko/20100101 Firefox/7.0.1", "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:2.0b4pre) Gecko/20100815 Minefield/4.0b4pre", "Mozilla/4.0 (compatible; MSIE 5.5; Windows NT 5.0 )", "Mozilla/4.0 (compatible; MSIE 5.5; Windows 98; Win 9x 4.90)", "Mozilla/5.0 (Windows; U; Windows XP) Gecko MultiZilla/1.6.1.0a", ]在 spider 同级目录下建立一个 MidWare 文件价里面写一个 HeaderMidWare.py 文件 内容为 # encoding: utf-8 from scrapy.utils.project import get_project_settings import random settings = get_project_settings() class ProcessHeaderMidware(): """process request add request info""" def process_request(self, request, spider): """ 随机从列表中获得header， 并传给user_agent进行使用 """ ua = random.choice(settings.get('USER_AGENT_LIST')) spider.logger.info(msg='now entring download midware') if ua: request.headers['User-Agent'] = ua # Add desired logging message here. spider.logger.info(u'User-Agent is : {} {}'.format(request.headers.get('User-Agent'), request)) pass在 setting.py 文件中添加 DOWNLOADER_MIDDLEWARES = { 'projectName.MidWare.HeaderMidWare.ProcessHeaderMidware': 543, } 方法 2：使用 fake_userAgent fake_userAgent github：https://github.com/sea1234/fake-useragent 安装 fake_userAgent：pip install fake-useragent from fake_useragent import UserAgent import requests ua = UserAgent() print(ua.ie) #ie浏览器的user agent print(ua.opera) #opera浏览器 print(ua.chrome) #chrome浏览器 print(ua.firefox) #firefox浏览器 print(ua.safari) #safri浏览器 #最常用的方式 #写爬虫最实用的是可以随意变换headers，一定要有随机性。支持随机生成请求头 print(ua.random) print(ua.random) print(ua.random) ##################################################### #请求的网址 url="http://www.baidu.com" #请求头 headers={"User-Agent":ua.random} #请求网址 response=requests.get(url=url,headers=headers) #响应体内容 print(response.text) #响应状态信息 print(response.status_code) #响应头信息 print(response.headers)user_agent_middlewares.py # -*- coding: utf-8 -*- from fake_useragent import UserAgent class RandomUserAgentMiddlware(object): #随机跟换user-agent def __init__(self,crawler): super(RandomUserAgentMiddlware,self).__init__() self.ua = UserAgent() self.ua_type = crawler.settings.get('RANDOM_UA_TYPE','random')#从setting文件中读取RANDOM_UA_TYPE值 @classmethod def from_crawler(cls,crawler): return cls(crawler) def process_request(self,request,spider): ###系统电泳函数 def get_ua(): return getattr(self.ua,self.ua_type) # user_agent_random=get_ua() request.headers.setdefault('User_Agent',get_ua()) pass在 setting.py 中添加 RANDOM_UA_TYPE = 'random' ##random chrome DOWNLOADER_MIDDLEWARES = { 'projectName.MidWare.user_agent_middlewares.RandomUserAgentMiddlware': 543, 'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware':None, } cookie 池基于 Scrapy-Redis 的分布式以及 cookies 池：https://cuiqingcai.com/4048.html 爬虫 中间件官网文档：https://docs.scrapy.org/en/latest/topics/spider-middleware.html 编写自己的爬虫中间件需要继承 scrapy.spidermiddlewares.SpiderMiddleware 基类，并重写以下几个方法。 process_spider_input(response, spider)process_spider_output(response, result, spider)process_spider_exception(response, exception, spider)process_start_requests(start_requests, spider)内置的爬虫中间件 scrapy 内置了5个爬虫中间件。
这里仅介绍一两个。 DepthMiddleware：该中间件记录了爬虫爬取请求地址的深度。我们可以使用DEPTH_LIMIT来指定爬虫爬取的深度。UrlLengthMiddleware：该中间件会过滤掉超出最大允许长度的URL，爬虫不会访问这些超长URL。最大长度通过URLLENGTH_LIMIT配置来指定，默认值是2083。URLLENGTH_LIMIT = 2083 内建服务 ( 日志、邮件、web服务 )scrapy 学习笔记(1、2、3) 快速入门：https://www.imooc.com/article/21838进阶使用：https://www.imooc.com/article/21839实战练习：https://www.imooc.com/article/21840scrapy内置了几个服务，可以让我们使用scrapy更加方便。 日志：爬虫类定义了 log 函数，我们可以方便的在爬虫类中记录日志。 import scrapy class MySpider(scrapy.Spider): name = 'myspider' start_urls = ['https://scrapinghub.com'] def parse(self, response): self.logger.info('Parse function called on %s', response.url)日志相关的配置，点击可以跳转到官方文档查看详细信息。 LOG_FILELOG_ENABLEDLOG_ENCODINGLOG_LEVELLOG_FORMATLOG_DATEFORMATLOG_STDOUTLOG_SHORT_NAMES发送电子邮件：有时候我们可能希望爬到一定数量的数据就发送电子邮件进行提醒。scrapy也内置了这个功能。我们可以通过构造函数参数来创建邮件发送器。 from scrapy.mail import MailSender mailer = MailSender(这里是构造函数参数) 也可以从配置文件实例化。 mailer = MailSender.from_settings(settings) 然后调用send方法就可以发送邮件了。 mailer.send( to=["[email protected]"], subject="Some subject", body="Some body", cc=["[email protected]"] ) 电子邮件相关配置参考官方文档。 web 服务：这个功能本来是写在官方文档内建服务条目下的，但是实际上这个功能已经变成了一个单独的项目，需要额外安装。pip install scrapy-jsonrpc 然后在扩展中包含这个功能。 EXTENSIONS = { 'scrapy_jsonrpc.webservice.WebService': 500, } 还需要在配置中启用该功能。 JSONRPC_ENABLED = True 然后在爬虫运行的时候访问 http://localhost:6080/crawler 即可查看爬虫运行情况了。 该项目的其他配置查看其官方文档。 优化爬虫 ( 增大并发数、增大线程池、降低日志级别 等)爬虫项目可以通过修改一些配置进行优化。 设置禁止跳转（code=301、302）、请求超时 DOWNLOAD_TIMEOUT = 10 REDIRECT_ENABLED = False 增大并发数：并发数可以通过下面的配置进行设置。具体的并发数需要根据服务器的CPU等设置来进行更改。一般来说服务器CPU使用在80%-90%之间利用率比较高。我们可以从并发数100开始反复进行测试。 CONCURRENT_REQUESTS = 100 增大线程池：scrapy 通过一个线程池来进行 DNS 查询，增大这个线程池一般也可以提高 scrapy 性能。 REACTOR_THREADPOOL_MAXSIZE = 20 降低日志级别：默认情况下scrapy使用debug级别来打印日志，通过降低日志级别，我们可以减少日志打印，从而提高程序运行速度。 LOG_LEVEL = 'INFO' 禁用 Cookie：如果不是必须的，我们可以通过禁用Cookie来提高性能。如果需要登录用户才能爬取数据，不要禁用Cookie。 COOKIES_ENABLED = False 关闭重试：频繁重试可能导致目标服务器响应缓慢，我们自己访问不了别人也访问不了。所以可以考虑关闭重试。 RETRY_ENABLED = False 减少下载超时：如果网络连接比较快的话，我们可以减少下载超时，让爬虫卡住的请求中跳出来，一般可以提高爬虫效率。 DOWNLOAD_TIMEOUT = 15 关闭重定向：如果不是必要的话，我们可以关闭重定向来提高爬虫性能。 REDIRECT_ENABLED = False 关掉重定向,不会重定向到新的地址 HTTPERROR_ALLOWED_CODES = [302,] 返回302时,按正常返回对待,可以正常写入cookie 或者启用下载中间件 #'scrapy.contrib.downloadermiddleware.redirect.RedirectMiddleware': 600, 自动调整爬虫负载：scrapy有一个扩展可以自动调节服务器负载，它通过一个算法来确定最佳的爬虫延时等设置。它的文档在这里。 相关配置如下，点击链接可以跳转到对应文档。 AUTOTHROTTLE_ENABLEDAUTOTHROTTLE_START_DELAYAUTOTHROTTLE_MAX_DELAYAUTOTHROTTLE_TARGET_CONCURRENCYAUTOTHROTTLE_DEBUGCONCURRENT_REQUESTS_PER_DOMAINCONCURRENT_REQUESTS_PER_IPDOWNLOAD_DELAY 编辑器中的调试在页面的任意位置添加如下代码 from scrapy.shell import inspect_response def paser(self,response): inspect_response(response,self) #当程序运行到这里就会跳出终端，并且在终端出现调试命令，当然这个可以随便写在哪里 暂停、恢复 爬虫初学者最头疼的事情就是没有处理好异常，当爬虫爬到一半的时候突然因为错误而中断了，但是这时又不能从中断的地方开始继续爬，顿时感觉心里日了狗，但是这里有一个方法可以暂时的存储你爬的状态，当爬虫中断的时候继续打开后依然可以从中断的地方爬，不过虽说持久化可以有效的处理，但是要注意的是当使用cookie临时的模拟登录状态的时候要注意cookie的有效期 只需要在setting.py中JOB_DIR=file_name 其中填的是你的文件目录，注意这里的目录不允许共享，只能存储单独的一个spdire的运行状态，如果你不想在从中断的地方开始运行，只需要将这个文件夹删除即可 当然还有其他的放法：scrapy crawl somespider -s JOBDIR=crawls/somespider-1，这个是在终端启动爬虫的时候调用的，可以通过ctr+c中断，恢复还是输入上面的命令 部署爬虫官方文档介绍了两种部署爬虫的方式，可以将爬虫部署到服务器上远程执行。第一种是通过Scrapyd开源项目来部署，也是这里要介绍的方式。第二种是通过scrapy公司提供的商业收费版服务Scrapy Cloud部署，推荐有财力的公司考虑。 服务器端：首先服务器需要安装scrapyd包，如果是Linux系统还可以考虑使用对应的包管理器来安装。 pip install scrapyd apt-get install scrapyd 然后运行scrapyd服务，如果使用系统包管理器安装，那么可能已经配置好了systemd文件。 scrapyd # 或者 systemctl enable scrapyd scrapyd 附带了一个简单的 web 界面可以帮助我们查看爬虫运行情况，默认情况下访问http://localhost:6800/ 来查看这个界面。 scrapyd 的配置文件可以是~/.scrapyd.conf或者/etc/scrapyd/scrapyd.conf。下面是一个简单配置，绑定所有端口，这样一来从任意位置都可以访问web界面。 [scrapyd] bind_address = 0.0.0.0 scrapyd的功能可以查看其 API文档。 客户端：客户端如果要上传爬虫，可以通过服务器API的端点addversion.json来实现，或者安装一个简便工具scrapyd-client。 首先安装客户端工具：pip install scrapyd-client 这个客户端目前好像有bug，在windows下运行scrapy-deploy命令不会直接执行，而是弹出一个文件关联对话框。如果你遇到这种情况，可以找到Python安装路径下的脚本路径（例如C:\Program Files\Python36\Scripts），然后编写一个scrapyd-deploy.bat批处理文件，内容如下。这样就可以正常运行了。 @"c:\program files\python36\python.exe" "c:\program files\python36\Scripts\scrapyd-deploy" %* 然后切换到项目路径，编辑项目全局配置文件scrapy.cfg，添加部署路径。 [deploy] url = http://192.168.64.136:6800/ project = quotesbot 然后直接运行scrapy-deploy命令，就可以看到项目已经成功部署到服务器上了。 运行爬虫需要使用scrapyd的API，例如使用curl，可以用下面的命令。 curl http://192.168.64.136:6800/schedule.json -d project=quotesbot -d spider=toscrape-css 或者使用Jetbrains 系列IDE 2017.3的基于编辑器的HTTP客户端。 然后点击Jobs就可以看到爬虫已经开始运行了。如果要查看状态，点击右边的log即可。 以上就是scrapy的进阶介绍了，利用这些功能，我们可以编写更加实用的爬虫，并将它们部署到服务器上。 URL 过滤 ( Bloom Filter、Hyperloglog )正常业务逻辑下，爬虫不会对重复请求爬取两次。所以爬虫默认都会对重复请求进行过滤，但当爬虫体量达到千万级时，默认的过滤器占用的内存将会远远超乎你的想象。 为了解决这个问题，可以通过一些算法来牺牲一点点过滤的准确性来换取更小的空间复杂度 Bloom Filter：可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都远远超过一般的算法，缺点是有一定的误识别率和删除困难。Hyperloglog：是一个基数估计算法。其空间效率非常高，1.5K内存可以在误差不超过2%的前提下，用于超过10亿的数据集合基数估计。这两种算法都是合适的选择，以 Hyperloglog 为例：由于 redis 已经提供了支持 hyperloglog 的数据结构，所以只需对此数据结构进行操作即可 基于 Scrapy-redis 的分布式爬虫设计：https://www.jianshu.com/p/cd4054bbc757/ # coding: utf-8 from scrapy import Item, Field from scrapy.spiders import Rule from scrapy_redis.spiders import RedisCrawlSpider from scrapy.linkextractors import LinkExtractor from redis import Redis from time import time from urllib.parse import urlparse, parse_qs, urlencode class MasterSpider(RedisCrawlSpider): name = 'ebay_master' redis_key = 'ebay:start_urls' ebay_main_lx = LinkExtractor(allow=(r'http://www.ebay.com/sch/allcategories/all-categories', )) ebay_category2_lx = LinkExtractor(allow=(r'http://www.ebay.com/sch/[^\s]*/\d+/i.html', r'http://www.ebay.com/sch/[^\s]*/\d+/i.html?_ipg=\d+&_pgn=\d+', r'http://www.ebay.com/sch/[^\s]*/\d+/i.html?_pgn=\d+&_ipg=\d+',)) rules = ( Rule(ebay_category2_lx, callback='parse_category2', follow=False), Rule(ebay_main_lx, callback='parse_main', follow=False), ) def __init__(self, *args, **kwargs): domain = kwargs.pop('domain', '') # self.allowed_domains = filter(None, domain.split(',')) super(MasterSpider, self).__init__(*args, **kwargs) def parse_main(self, response): pass data = response.xpath("//div[@class='gcma']/ul/li/a[@class='ch']") for d in data: try: item = LinkItem() item['name'] = d.xpath("text()").extract_first() item['link'] = d.xpath("@href").extract_first() yield self.make_requests_from_url(item['link'] + r"?_fsrp=1&_pppn=r1&scp=ce2") except: pass def parse_category2(self, response): data = response.xpath("//ul[@id='ListViewInner']/li/h3[@class='lvtitle']/a[@class='vip']") redis = Redis() for d in data: # item = LinkItem() try: self._filter_url(redis, d.xpath("@href").extract_first()) except: pass try: next_page = response.xpath("//a[@class='gspr next']/@href").extract_first() except: pass else: # yield self.make_requests_from_url(next_page) new_url = self._build_url(response.url) redis.lpush("test:new_url", new_url) # yield self.make_requests_from_url(new_url) # yield Request(url, headers=self.headers, callback=self.parse2) def _filter_url(self, redis, url, key="ebay_slave:start_urls"): is_new_url = bool(redis.pfadd(key + "_filter", url)) if is_new_url: redis.lpush(key, url) def _build_url(self, url): parse = urlparse(url) query = parse_qs(parse.query) base = parse.scheme + '://' + parse.netloc + parse.path if '_ipg' not in query.keys() or '_pgn' not in query.keys() or '_skc' in query.keys(): new_url = base + "?" + urlencode({"_ipg": "200", "_pgn": "1"}) else: new_url = base + "?" + urlencode({"_ipg": query['_ipg'][0], "_pgn": int(query['_pgn'][0]) + 1}) return new_url class LinkItem(Item): name = Field() link = Field()当 redis.pfadd() 执行时，一个 url 尝试插入 hyperloglog 结构中，如果 url 存在返回 0，反之返回 1。由此来判断是否要将该 url 存放至待爬队列 集成 bloomfilter 到 scrapy-redis 中 scrapy_redis去重优化（已有7亿条数据）：https://blog.csdn.net/bone_ace/article/details/53099042 将bloomfilter(布隆过滤器)集成到scrapy-redis中：https://www.cnblogs.com/adc8868/p/7442306.html 模拟用户登录有时候需要模拟用户登录，这时候可以使用 FormRequest.from_response 方法。这时候爬虫功能稍有变化，parse 函数用来发送用户名和密码，抽取数据的操作放在回调函数中进行。 import scrapy class LoginSpider(scrapy.Spider): name = 'example.com' start_urls = ['http://www.example.com/users/login.php'] def parse(self, response): return scrapy.FormRequest.from_response( response, formdata={'username': 'john', 'password': 'secret'}, callback=self.after_login ) def after_login(self, response): # 检查是否登录成功 if "authentication failed" in response.body: self.logger.error("Login failed") return # 在这里继续爬取数据模拟登陆过程中遇到的那些坑：https://blog.csdn.net/amaomao123/article/details/52511882 动态 JavaScript 渲染、无头浏览器 playwright-python：https://docs.scrapy.org/en/latest/topics/dynamic-content.html
scrapy_redis_cluster （ 已经不在维护 ）：https://github.com/thsheep/scrapy_redis_cluster scrapy-redis-cluster ：https://pypi.org/project/scrapy-redis-cluster scrapy-redis-cluster 已经不在维护 ！！！！！ scrapyd-redis 的集群版 此包Python名称：scrapy-redis-cluster目前版本： scrapy-redis-cluster 0.4最后维护时间：Jul 5, 2018摘要：scrapyd-redis的集群版安装命令：pip install scrapy-redis-cluster其它：scrapy-redis-cluster 这个Python第三方库的作者没有提供更多的项目描述信息了，2019-11-10 23:44:14。 scrapy-redis 使用 redis 集群进行分布式爬取正常情况单机的redis可以满足scrapy-redis进行分布式爬取，可是如果单机的redis的内存过小，很容易导致系统内存不够，读取数据缓慢，如果使用docker运行redis，更加可能导致redis的容器的进程被杀掉。（笔者就曾经经常遇到这种情况，机器内存才8GB，上面跑了N个docker容器，一旦内存吃紧，某个容器就被kill掉，导致爬虫经常出问题）。 使用 redis 集群可以增加 redis 集体内存，防止出现上面的情况。 scrapy redis-cluster 很简单，只需要按照以下步骤： 1. 安装库：pip install scrapy-redis-cluster 2. 修改 settings 文件 # Redis集群地址 REDIS_MASTER_NODES = [ {"host": "192.168.10.233", "port": "30001"}, {"host": "192.168.10.234", "port": "30002"}, {"host": "192.168.10.235", "port": "30003"}, ] # 使用的哈希函数数，默认为6 BLOOMFILTER_HASH_NUMBER = 6 # Bloomfilter使用的Redis内存位，30表示2 ^ 30 = 128MB，默认为22 (1MB 可去重130W URL) BLOOMFILTER_BIT = 22 # 不清空redis队列 SCHEDULER_PERSIST = True # 调度队列 SCHEDULER = "scrapy_redis_cluster.scheduler.Scheduler" # 去重 DUPEFILTER_CLASS = "scrapy_redis_cluster.dupefilter.RFPDupeFilter" # queue SCHEDULER_QUEUE_CLASS = 'scrapy_redis_cluster.queue.PriorityQueue' 3、scrapy-redis-sentinel（ 集群版_2 ）scrapy-redis-sentinel ：https://github.com/crawlaio/scrapy-redis-sentinel pypi 地址：scrapy-redis-sentinel · PyPI 基于原项目 scrpy-redis：https://github.com/rmax/scrapy-redis 进行修改，修改内容如下： 添加了 Redis 哨兵连接支持添加了 Redis 集群连接支持添加了 Bloomfilter 去重安装第三方库：pip install scrapy-redis-sentinel 原版本 scrpy-redis 的所有配置都支持。优先级：哨兵模式 > 集群模式 > 单机模式 配置示例 # ----------------------------------------Bloomfilter 配置------------------------------------- # 使用的哈希函数数，默认为 6 BLOOMFILTER_HASH_NUMBER = 6 # Bloomfilter 使用的 Redis 内存位，30 表示 2 ^ 30 = 128MB，默认为 30 (2 ^ 22 = 1MB 可去重 130W URL) BLOOMFILTER_BIT = 30 # 是否开启去重调试模式 默认为 False 关闭 DUPEFILTER_DEBUG = False # ----------------------------------------Redis 单机模式------------------------------------- # Redis 单机地址 REDIS_HOST = "172.25.2.25" REDIS_PORT = 6379 # REDIS 单机模式配置参数 REDIS_PARAMS = { "password": "password", "db": 0 } # ----------------------------------------Redis 哨兵模式------------------------------------- # Redis 哨兵地址 REDIS_SENTINELS = [ ('172.25.2.25', 26379), ('172.25.2.26', 26379), ('172.25.2.27', 26379) ] # REDIS_SENTINEL_PARAMS 哨兵模式配置参数。 REDIS_SENTINEL_PARAMS= { "service_name":"mymaster", "password": "password", "db": 0 } # ----------------------------------------Redis 集群模式------------------------------------- # Redis 集群地址 REDIS_STARTUP_NODES = [ {"host": "172.25.2.25", "port": "6379"}, {"host": "172.25.2.26", "port": "6379"}, {"host": "172.25.2.27", "port": "6379"}, ] # REDIS_CLUSTER_PARAMS 集群模式配置参数 REDIS_CLUSTER_PARAMS= { "password": "password" } # ----------------------------------------Scrapy 其他参数------------------------------------- # 在 redis 中保持 scrapy-redis 用到的各个队列，从而允许暂停和暂停后恢复，也就是不清理 redis queues SCHEDULER_PERSIST = True # 调度队列 SCHEDULER = "scrapy_redis_sentinel.scheduler.Scheduler" # 去重 DUPEFILTER_CLASS = "scrapy_redis_sentinel.dupefilter.RFPDupeFilter" # 指定排序爬取地址时使用的队列 # 默认的 按优先级排序( Scrapy 默认)，由 sorted set 实现的一种非 FIFO、LIFO 方式。 # SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderPriorityQueue' # 可选的 按先进先出排序（FIFO） # SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderStack' # 可选的 按后进先出排序（LIFO） # SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderStack'注：当使用集群时单机不生效 spiders 使用原版本 scrpy-redis 使用方式 from scrapy_redis.spiders import RedisSpider class Spider(RedisSpider): ...修改 RedisSpider 引入方式后，scrapy-redis-sentinel 的使用方式 from scrapy_redis_sentinel.spiders import RedisSpider class Spider(RedisSpider): ... 使用示例：修改 setting.py文件 ITEM_PIPELINES = { 'scrapy_redis_sentinel.pipelines.RedisPipeline': 543, } # Bloomfilter 配置 # 使用的哈希函数数，默认为 6 BLOOMFILTER_HASH_NUMBER = 6 # Bloomfilter 使用的 Redis 内存位，30 表示 2 ^ 30 = 128MB，默认为 30 (2 ^ 22 = 1MB 可去重 130W URL) BLOOMFILTER_BIT = 30 # 是否开启去重调试模式 默认为 False 关闭 DUPEFILTER_DEBUG = False # Redis 集群地址 REDIS_MASTER_NODES = [ {"host": "192.168.56.30", "port": "9000"}, {"host": "192.168.56.31", "port": "9000"}, {"host": "192.168.56.32", "port": "9000"}, ] # REDIS_CLUSTER_PARAMS 集群模式配置参数 REDIS_CLUSTER_PARAMS= { # "password": "password" } # scrapy其他参数 # 在 redis 中保持 scrapy-redis 用到的各个队列，从而允许暂停和暂停后恢复，也就是不清理 redis queues SCHEDULER_PERSIST = True # 调度队列 SCHEDULER = "scrapy_redis_sentinel.scheduler.Scheduler" # 去重 DUPEFILTER_CLASS = "scrapy_redis_sentinel.dupefilter.RFPDupeFilter" # 指定排序爬取地址时使用的队列 # 默认的 按优先级排序( Scrapy 默认)，由 sorted set 实现的一种非 FIFO、LIFO 方式。 SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderPriorityQueue' # 可选的 按先进先出排序（FIFO） # SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderStack' # 可选的 按后进先出排序（LIFO） # SCHEDULER_QUEUE_CLASS = 'scrapy_redis_sentinel.queue.SpiderStack'修改 spider from scrapy_redis_sentinel.spiders import RedisSpider class scrapy_spider(RedisSpider): ......Redis 集群（ Redis5.0.7集群搭建：https://blog.csdn.net/pcengineercn/article/details/104502061 ） 经过调试，修复了一个bug（使用默认爬取队列时会报错），需要将源码中的 PriorityQueue（位于 Python 安装目录 /lib/python3.6/site-packages/scrapy_redis_sentinel/queue.py）替换为如下 class PriorityQueue(Base): """Per-spider priority queue abstraction using redis' sorted set""" def __len__(self): """Return the length of the queue""" return self.server.zcard(self.key) def push(self, request): """Push a request""" data = self._encode_request(request) score = -request.priority # We don't use zadd method as the order of arguments change depending on # whether the class is Redis or StrictRedis, and the option of using # kwargs only accepts strings, not bytes. self.server.execute_command("ZADD", self.key, score, data) def pop(self, timeout=0): """ Pop a request timeout not support in this queue class """ if not isinstance(self.server, RedisCluster): # use atomic range/remove using multi/exec pipe = self.server.pipeline() pipe.multi() pipe.zrange(self.key, 0, 0).zremrangebyrank(self.key, 0, 0) results, count = pipe.execute() if results: return self._decode_request(results[0]) # 使用集群的时候不能使用 multi/exec 来完成一个事务操作；使用lua脚本来实现类似功能 pop_lua_script = """ local result = redis.call('zrange', KEYS[1], 0, 0) local element = result[1] if element then redis.call('zremrangebyrank', KEYS[1], 0, 0) return element else return nil end """ script = self.server.register_script(pop_lua_script) results = script(keys=[self.key]) if results: return self._decode_request(results) |
【本文地址】
| 今日新闻 |
| 推荐新闻 |
| 专题文章 |