Python爬虫之Scrapy框架系列（24）——分布式爬虫scrapy_redis完整实战【XXTop250完整爬取】

目录：

  • 1.使用分布式爬取XX电影信息
    • （1）settings.py文件中的配置：
    • （2）spider文件的更改：
    • （3）items.py文件（两个项目一致！）：
    • （4）pipelines.py文件：
    • 分布式实现效果：
      • ①直接运行项目，发现在等待：
      • ②再开一个终端，做如下操作：
    • 总结：
    • 效果：
  • 2.解决一些小问题：
    • 2.1 解决爬空问题：（在两个项目中都进行以下操作！）
      • ①使用拓展程序（这个文件就是为了解决爬空而生的）：
      • ②在settings.py文件中设置这个拓展程序：
  • 3. 关于分布式（Scrapy\_redis）的总结：

1.使用分布式爬取XX电影信息

  • （此处做了限制，只爬取四页电影数据共计100条，可去除限制爬取全部10页250条数据！）

**项目源码：
链接：https://pan.baidu.com/s/13akXDxNbtBeRTUzUB_2SNQ
提取码：bcuy
**

目标：在本机上使用两个完全一模一样的豆瓣项目，去使用分布式下载XX电影top250电影信息！
在这里插入图片描述
其实，我们要进行修改的就只有settings.py文件以及爬虫文件，别的文件都不需要进行改动。

（1）settings.py文件中的配置：

  • （两个项目都做此配置）
#设置scrapy-redis
#1.启用调度将请求存储进redis
from scrapy_redis.scheduler import Scheduler
SCHEDULER="scrapy_redis.scheduler.Scheduler"#2.确保所有spider通过redis共享相同的重复过滤
from scrapy_redis.dupefilter import RFPDupeFilter
DUPEFILTER_CLASS="scrapy_redis.dupefilter.RFPDupeFilter"#3.指定连接到Redis时要使用的主机和端口     目的是连接上redis数据库
REDIS_HOST="localhost"
REDIS_PORT=6379# 不清理redis队列，允许暂停/恢复抓取    （可选）    允许暂停，redis数据不丢失     可以实现断点续爬！！！
SCHEDULER_PERSIST = True# 第二步：开启将数据存储进redis公共区域的管道！
# Configure item pipelines
# See https://docs.scrapy.org/en/latest/topics/item-pipeline.html
ITEM_PIPELINES = {'scrapy_redis.pipelines.RedisPipeline': 100,    # 开启数据交给redis公共区域的管道'douban.pipelines.DoubanPipeline': 200,         # 存储本地txt文件的管道
}

（2）spider文件的更改：

  • （两个项目略有不同！）

总共四步：

  1. 导入RedisSpider类：（既然要使用它，肯定首先要导入！）
    from scrapy_redis.spiders import RedisSpider

  2. 继承使用RedisSpider类：（既然要使用它，就要继承去使用这个类）
    class DbSpider(RedisSpider):

  3. 既然将请求都放进了Redis里，那爬虫文件中就不再需要start_urls这个初始请求了：
    #start_urls = ['https://movie.douban.com/top250']

  4. 设置一个键，寻找起始的url：（这个键就会在redis中寻找初始的url，所以后面我们只需往redis里放请求即可!）
    redis_key="db:start_urls"

完整版爬虫文件：
第一个项目下的爬虫文件：

# -*- coding: utf-8 -*-
import scrapy
import refrom ..items import DoubanItemfrom scrapy_redis.spiders import RedisSpider        # 1.导出RedisSpider类class DbSpider(RedisSpider):                        # 2.使用RedisSpider类name = 'db'allowed_domains = ['movie.douban.com']# start_urls = ['https://movie.douban.com/top250']   # 3.将请求放进redis里redis_key = "db:start_urls"                          # 4.设置一个键，寻找起始的urlpage_num = 0  # 类变量def parse(self, response):  # 解析和提取数据print('第一个项目：', response.url)print('第一个项目：', response.url)print('第一个项目：', response.url)# 获取电影信息数据# films_name=response.xpath('//div[@class="info"]/div/a/span[1]/text()').extract()node_list = response.xpath('//div[@class="info"]')  # 25个if node_list:  # 此判断的作用：在爬取到10页之后，就获取不到了！判断每次是否获取到数据，如果没有则返回空（即停止了）for node in node_list:# 电影名字film_name = node.xpath('./div/a/span[1]/text()').extract()[0]# 主演   拿标签内容，再正则表达式匹配con_star_name = node.xpath('./div/p[1]/text()').extract()[0]if "主" in con_star_name:star_name = re.findall("主演?:? ?(.*)", con_star_name)[0]else:star_name = "空"# 评分score = node_list.xpath('./div/div/span[@property="v:average"]/text()').extract()[0]# 使用字段名  收集数据item = DoubanItem()item["film_name"] = film_nameitem["star_name"] = star_nameitem["score"] = score# 形式：{"film_name":"肖申克的救赎","star_name":"蒂姆","score":"9.7"}detail_url = node.xpath('./div/a/@href').extract()[0]yield scrapy.Request(detail_url,callback=self.get_detail,meta={"info":item})# 此处几行的代码配合yield里传的参数meta={"num":self.page_num},共同作用实现：# 两个项目的共享变量page_num能正确变化，不导致冲突！！！if response.meta.get("num"):self.page_num = response.meta["num"]self.page_num += 1if self.page_num == 4:returnprint("page_num:", self.page_num)page_url = "https://movie.douban.com/top250?start={}&filter=".format(self.page_num * 25)yield scrapy.Request(page_url, callback=self.parse, meta={"num": self.page_num})# 注意：各个模块的请求都会交给引擎，然后经过引擎的一系列操作；但是，切记：引擎最后要把得到的数据再来给到# spider爬虫文件让它解析并获取到真正想要的数据（callback=self.parse)这样就可以再给到自身。else:returndef get_detail(self, response):item = DoubanItem()# 获取电影简介信息# 1.meta会跟随response一块返回  2.可以通过response.meta接收   3.通过updata可以添加到新的item对象info = response.meta["info"]  # 接收电影的基本信息item.update(info)  # 把电影基本信息的字段加进去# 将电影简介信息加入相应的字段里description = response.xpath('//div[@id="link-report-intra"]//span[@property="v:summary"]/text()').extract()[0]\.strip()item['description'] = descriptionyield item

第二个项目下的爬虫文件：

# -*- coding: utf-8 -*-
import scrapy
import refrom ..items import DoubanItemfrom scrapy_redis.spiders import RedisSpider        # 1.导出RedisSpider类class DbSpider(RedisSpider):                        # 2.使用RedisSpider类name = 'db'allowed_domains = ['movie.douban.com']# start_urls = ['https://movie.douban.com/top250']   # 3.将请求放进redis里redis_key = "db:start_urls"                          # 4.设置一个键，寻找起始的urlpage_num = 0  # 类变量def parse(self, response):  # 解析和提取数据print('第二个项目：', response.url)print('第二个项目：', response.url)print('第二个项目：', response.url)# 获取电影信息数据# films_name=response.xpath('//div[@class="info"]/div/a/span[1]/text()').extract()node_list = response.xpath('//div[@class="info"]')  # 25个if node_list:  # 此判断的作用：在爬取到10页之后，就获取不到了！判断每次是否获取到数据，如果没有则返回空（即停止了）for node in node_list:# 电影名字film_name = node.xpath('./div/a/span[1]/text()').extract()[0]# 主演   拿标签内容，再正则表达式匹配con_star_name = node.xpath('./div/p[1]/text()').extract()[0]if "主" in con_star_name:star_name = re.findall("主演?:? ?(.*)", con_star_name)[0]else:star_name = "空"# 评分score = node_list.xpath('./div/div/span[@property="v:average"]/text()').extract()[0]# 使用字段名  收集数据item = DoubanItem()item["film_name"] = film_nameitem["star_name"] = star_nameitem["score"] = score# 形式：{"film_name":"肖申克的救赎","star_name":"蒂姆","score":"9.7"}detail_url = node.xpath('./div/a/@href').extract()[0]yield scrapy.Request(detail_url,callback=self.get_detail,meta={"info":item})# 此处几行的代码配合57行yield里传的参数meta={"num":self.page_num},共同作用实现：# 两个项目的共享变量page_num能正确变化，不导致冲突！！！if response.meta.get("num"):self.page_num = response.meta["num"]self.page_num += 1if self.page_num == 4:returnprint("page_num:", self.page_num)page_url = "https://movie.douban.com/top250?start={}&filter=".format(self.page_num * 25)yield scrapy.Request(page_url, callback=self.parse, meta={"num": self.page_num})# 注意：各个模块的请求都会交给引擎，然后经过引擎的一系列操作；但是，切记：引擎最后要把得到的数据再来给到# spider爬虫文件让它解析并获取到真正想要的数据（callback=self.parse)这样就可以再给到自身。else:returndef get_detail(self, response):item = DoubanItem()# 获取电影简介信息# 1.meta会跟随response一块返回  2.可以通过response.meta接收   3.通过updata可以添加到新的item对象info = response.meta["info"]  # 接收电影的基本信息item.update(info)  # 把电影基本信息的字段加进去# 将电影简介信息加入相应的字段里description = response.xpath('//div[@id="link-report-intra"]//span[@property="v:summary"]/text()').extract()[0]\.strip()item['description'] = descriptionyield item

（3）items.py文件（两个项目一致！）：

# -*- coding: utf-8 -*-# Define here the models for your scraped items
#
# See documentation in:
# https://docs.scrapy.org/en/latest/topics/items.htmlimport scrapyclass DoubanItem(scrapy.Item):# define the fields for your item here like:# name = scrapy.Field()#需要定义字段名  就像数据库那样，有字段名，才能插入数据（即存储数据）# films_name=scrapy.Field()   #定义字段名film_name=scrapy.Field()star_name=scrapy.Field()score=scrapy.Field()description = scrapy.Field()

（4）pipelines.py文件：

  • （两个项目存储本地txt文件名可改为不一样的，便于观察！）
# -*- coding: utf-8 -*-# Define your item pipelines here
#
# Don't forget to add your pipeline to the ITEM_PIPELINES setting
# See: https://docs.scrapy.org/en/latest/topics/item-pipeline.htmlimport json
import pymysqlclass DoubanPipeline(object):def open_spider(self,spider):   #爬虫文件开启，此方法就开启self.f=open("films.txt","w",encoding="utf-8")       #打开文件def process_item(self, item, spider):        #会来25次，就会调用25次这个方法  如果按常规来写，文件就会被操作25次打开关闭#为了能写进text  json.dumps将dic数据转换为strjson_str=json.dumps(dict(item),ensure_ascii=False)+"\n"self.f.write(json_str)                              #爬虫文件开启时，文件就已经打开，在此直接写入数据即可！return itemdef close_spider(self,spider):  #爬虫文件关闭，此方法就开启self.f.close()                                      #爬虫文件关闭时，引擎已经将全部数据交给管道，关闭文件

分布式实现效果：

①直接运行项目，发现在等待：

分别在两个终端中开启两个scrapy项目：（注意：之前要开启redis数据库）

在这里插入图片描述
会发现，这俩项目都在等待，不会继续执行。这是因为没有给redis这个公共区域一个初始的请求，这俩项目都在周而复始的向redis要初始url，结果一直要不到！

在两个项目的settings.py文件中设置两个的日志不显示在控制台，而是存储到.log文件中。为了便于观察：

LOG_FILE="db.log"
LOG_ENABLED=False

②再开一个终端，做如下操作：

lpush db:start_urls https://movie.douban.com/top250

在这里插入图片描述
会发现我们的两个项目都会成功的跑起来：（而且总共获取数据刚好是四页的电影信息，共计100条）

在这里插入图片描述在这里插入图片描述

总结：

会发现，第一个项目运行会显示使用了parse函数，这也就说明在redis这个公共区域的start_urls请求被第一个项目抢到了，然后就会运行这个项目，
但是，在这个项目的爬虫文件代码执行的过程中会在25次循环中给引擎发送共25次url请求，引擎得到这25个request请求后会将它们都交给scheduler调度器，再通过调度器交给redis数据库这个公共区域。
然后，两个项目的scheduler调度器就会一起抢这公共区域里的请求，并在各自的爬虫程序运行过程中提交给redis别的请求，两个项目继续抢，直到爬空。这就实现了咱爬虫的分布式爬取数据！！！
在这里插入图片描述

效果：

  • （因为没有解决爬空，所以项目运行完并不会自己关闭，而且，哪怕项目运行完了，也会一直无限的爬空，就导致两个项目爬取的保存本地的数据不够100条，所以，在两个项目运行完在爬空的时候，强制关闭两个项目，就会发现数据是完整的了！！！）

两个项目下的获取存储到本地的txt文本内的电影信息共计刚好我们所要爬取的所有目标数据：四页共100部电影的信息。

2.解决一些小问题：

2.1 解决爬空问题：（在两个项目中都进行以下操作！）

①使用拓展程序（这个文件就是为了解决爬空而生的）：

两个项目进行防爬空设置后，如果数据爬取完成，在指定时间内就会自动停止爬虫！！！
（文件名：extensions.py，放到settings.py同级目录里）

加入此拓展之后完整的项目代码：
链接：https://pan.baidu.com/s/1Naie1HsWCxS-1ntorT3_RQ
提取码：e30p

# -*- coding: utf-8 -*-# Define here the models for your scraped Extensions
import loggingfrom scrapy import signals
from scrapy.exceptions import NotConfiguredlogging = logging.getLogger(__name__)class RedisSpiderSmartIdleClosedExensions(object):def __init__(self, idle_number, crawler):self.crawler = crawlerself.idle_number = idle_numberself.idle_list = []self.idle_count = 0@classmethoddef from_crawler(cls, crawler):# first check if the extension should be enabled and raise# NotConfigured otherwiseif not crawler.settings.getbool('MYEXT_ENABLED'):raise NotConfiguredif not 'redis_key' in crawler.spidercls.__dict__.keys():raise NotConfigured('Only supports RedisSpider')# get the number of items from settingsidle_number = crawler.settings.getint('IDLE_NUMBER', 360)# instantiate the extension objectext = cls(idle_number, crawler)# connect the extension object to signalscrawler.signals.connect(ext.spider_opened, signal=signals.spider_opened)crawler.signals.connect(ext.spider_closed, signal=signals.spider_closed)crawler.signals.connect(ext.spider_idle, signal=signals.spider_idle)return extdef spider_opened(self, spider):spider.logger.info("opened spider {}, Allow waiting time:{} second".format(spider.name, self.idle_number * 5))def spider_closed(self, spider):spider.logger.info("closed spider {}, Waiting time exceeded {} second".format(spider.name, self.idle_number * 5))def spider_idle(self, spider):# 程序启动的时候会调用这个方法一次，之后每隔5秒再请求一次# 当持续半个小时都没有spider.redis_key，就关闭爬虫# 判断是否存在 redis_keyif not spider.server.exists(spider.redis_key):self.idle_count += 1else:self.idle_count = 0if self.idle_count > self.idle_number:# 执行关闭爬虫操作self.crawler.engine.close_spider(spider, 'Waiting time exceeded')

②在settings.py文件中设置这个拓展程序：

# Enable or disable extensions                  #扩展程序
# See https://docs.scrapy.org/en/latest/topics/extensions.html
EXTENSIONS = {# 'scrapy.extensions.telnet.TelnetConsole': None,'film.extensions.RedisSpiderSmartIdleClosedExensions':500,					#开启extensions.py这个拓展程序
}
MYEXT_ENABLED = True      # 开启扩展
IDLE_NUMBER = 3           # 配置空闲持续时间单位为 3个 ，一个时间单位为5s

注意：redis中存储的数据：

  • spidername:items
    list类型，保存爬虫获取到的数据item内容是json字符串。
  • spidername:dupefilter
    set类型，用于爬虫访问的URL去重内容是40个字符的url的hash字符串
  • spidername:start_urls
    list类型，用于接收redisspider启动时的第一个url
  • spidername:requests
    zset类型，用于存放requests等待调度。内容是requests对象的序列化字符串。

3. 关于分布式（Scrapy_redis）的总结：

(一)分布式爬虫
一.settings里的配置
# 启用调度将请求存储进redis
# 1.必须
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
#2. 必须
# 确保所有spider通过redis共享相同的重复过滤。
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"# 3.必须
# 指定连接到Redis时要使用的主机和端口。
REDIS_HOST = 'localhost'
REDIS_PORT = 6379二.spider文件更改from scrapy_redis.spiders import  RedisSpider #1 导出 RedisSpiderclass DbSpider(RedisSpider):  #2使用RedisSpider类# start_urls = ['https://movie.douban.com/top250/']  #3将要请求放在  公共区域 redis里面redis_key = "db:start_urls"#4  设置一个键  寻找起始url三.redis数据库中 写入  start_urls      
lpush  db:start_urls   https://movie.douban.com/top250/四.解决爬空的问题
1.解决爬空的文件    extensions.py  主要是RedisSpiderSmartIdleClosedExensions
2.设置
MYEXT_ENABLED = True      # 开启扩展
IDLE_NUMBER = 3           # 配置空闲持续时间单位为 3个 ，一个时间单位为5s

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/338965.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Hive分区表实战 - 单分区字段

Hive分区表实战 - 单分区字段

文章目录 一、实战概述二、实战步骤（一）创建图书数据库（二）创建国别分区的图书表（三）在本地创建数据文件（四）按分区加载数据1、加载中文书籍数据到countrycn分区2、加载英文书籍数据…
阅读更多...
7个JavaScript面试题全面解析,一文搞定技术面试

7个JavaScript面试题全面解析,一文搞定技术面试

JavaScript是构建网络的主要基石之一。这个强大的语言也有自己的怪癖。例如,您知道0 -0计算为true,或者Number("")产生0吗? 问题在于,这些怪癖有时会让你抓耳挠腮,甚至质疑Brendon Eich发明JavaScript的那一天是不是high了。当然,这里的重点不是说JavaScript是一种…
阅读更多...
RV1126边缘计算AI盒子，支持4-6路1080p视频，2T 算力

RV1126边缘计算AI盒子，支持4-6路1080p视频，2T 算力

1 产品概述 信迈推出基于瑞芯微Rockchip RV1126架构的AI边缘计算主板，RV1126芯片是四核ARM Cortex-A7,1.5GHz， RSIC-V 200MHz CPU ，NPU2.0Tops。AI边缘计算主板外围接口丰富，拥有超强扩展性，可广泛应用在智慧安防、工…
阅读更多...
统一密钥管理在信息安全领域有什么作用

统一密钥管理在信息安全领域有什么作用

统一密钥管理在信息安全领域中至关重要。它可以确保密钥的安全性、保密性和可用性，同时降低开发、维护和管理的成本。 对于没有KMS(密钥管理服务)管理系统的公司，密钥的本地化管理可能导致密钥分散在代码、配置文件中，缺乏统一管理&#xff0…
阅读更多...
机器学习笔记一之入门概念

机器学习笔记一之入门概念

目录 一 基本分类二 按模型分类概率模型（Probabilistic Models）非概率模型（Non-Probabilistic Models）对比结论线性模型 (Linear Models)非线性模型 (Non-linear Models)对比 三 按算法分类1.批量学习（Batch Learning&…
阅读更多...
HarmonyOS 容器组件(Column Row Flex)

HarmonyOS 容器组件(Column Row Flex)

今天 我们来说容器组件中的 Column Row Flex Column 我们应该比较熟了 之前用了很多了 是一个列容器 老规矩 先来一个组件骨架 Entry Component struct Index {build() {Column({space: 30}) {}.width(100%).height(100%)} }我们在中的 Column 元素中加入代码 Column() {Co…
阅读更多...
某集成电路中高端测试设备厂商：大幅提升网间文件交换效率

某集成电路中高端测试设备厂商：大幅提升网间文件交换效率

集成电路中高端测试设备厂商 该集成电路测试厂商是一家从事集成电路测试设备研发设计、制造、销售和服务的高科技企业，公司研发的中高端自动化测试设备产品填补了中国集成电路中高端测试设备领域的空白，改变目前完全依赖国外进口的现状同时本着与国外领…
阅读更多...
VS中打开ui文件闪退

VS中打开ui文件闪退

解决办法： 依次点击《扩展》-> 《Qt vs tools》-> 《options》-> 《Qt》-> 《general》 -> 《Qt Designer》 -> 《run in detached window》 -> true
阅读更多...
ffmpeg写YUV420文件碰到阶梯型横线或者条纹状画面的原因和解决办法

ffmpeg写YUV420文件碰到阶梯型横线或者条纹状画面的原因和解决办法

版权声明：本文为CSDN博主「文三~」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。 原文链接：https://blog.csdn.net/asdasfdgdhh/article/details/112831581 留作备份 阶梯型横线： 条纹状画面…
阅读更多...
Oracle regexp_substr

Oracle regexp_substr

select regexp_substr(123|456|789, [^|], 1, 2) from dual;
阅读更多...
【vitest 单元测试】如何蹭 ant-design-web3 的PR

【vitest 单元测试】如何蹭 ant-design-web3 的PR

这篇文章分享单测经验，希望你能收获到有用的单测知识或者pr思路，填补单测的过程可以深刻理解组件内部的每一个流程，相信一定有所收获。 ant-design-web3 前言查看单测覆盖情况运行命令，本地会生成一份临时目录通过live server打开…
阅读更多...
频率阈图像滤波

频率阈图像滤波

介绍 频率阈图像滤波是一种在频域中进行图像处理的方法，它基于图像的频率分布来实现滤波效果。具体步骤如下： 将原始图像转换到频域：使用快速傅里叶变换（FFT）将图像从空间域转换到频域。对频域图像应用频率阈滤波器&a…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023