前言
（备注一下，我的开发环境不是Linux就是MacOSX，Windows很多写法不是这样的）
在爬取数据的过程中，有时候需要用到定时、增量爬取。定时这里暂且不说，先说增量爬取。
- 我想要的增量爬取目前只是简单的，根据url请求来判断是否爬过，如果爬过则不再爬。
- 复杂一些的增量则是重复爬取，根据指定的几个字段判断是否值有变化，值有变化也算作增量，应当爬取且只更新变化部分（比如天猫商品数据，商品的价格有变化则更新价格，但是url是重复的，也应当爬取）
网上增量爬取的文章很多，包括看过慕课网Scrapy课的笔记,但是它还是不完善，我将在这个基础上进行实际集成。
布隆简介
Bloom Filter是一种空间效率很高的随机数据结构，它利用位数组很简洁地表示一个集合，并能判断一个元素是否属于这个集合。Bloom Filter的这种高效是有一定代价的：在判断一个元素是否属于某个集合时，有可能会把不属于这个集合的元素误认为属于这个集合（false positive）。因此，Bloom Filter不适合那些“零错误”的应用场合。而在能容忍低错误率的应用场合下，Bloom Filter通过极少的错误换取了存储空间的极大节省。
具体的bloomfilter概念和原理应该查看这篇文章：传送，还有《海量数据处理算法》以及《大规模数据处理利器》
布隆优点
相比于其它的数据结构，布隆过滤器在空间和时间方面都有巨大的优势。布隆过滤器存储空间和插入/查询时间都是常数。另外, Hash 函数相互之间没有关系，方便由硬件并行实现。布隆过滤器不需要存储元素本身，在某些对保密要求非常严格的场合有优势。
布隆过滤器可以表示全集，其它任何数据结构都不能；
k 和 m 相同，使用同一组 Hash 函数的两个布隆过滤器的交并差运算可以使用位操作进行。
布隆缺点
但是布隆过滤器的缺点和优点一样明显。误算率（False Positive）是其中之一。随着存入的元素数量增加，误算率随之增加。但是如果元素数量太少，则使用散列表足矣。
另外，一般情况下不能从布隆过滤器中删除元素. 我们很容易想到把位列阵变成整数数组，每插入一个元素相应的计数器加1, 这样删除元素时将计数器减掉就可以了。然而要保证安全的删除元素并非如此简单。首先我们必须保证删除的元素的确在布隆过滤器里面. 这一点单凭这个过滤器是无法保证的。另外计数器回绕也会造成问题。
总的来说，布隆很适合来处理海量的数据，而且速度优势很强。
redis与bloom
去重”是日常工作中会经常用到的一项技能，在爬虫领域更是常用，并且规模一般都比较大。参考文章《基于Redis的Bloomfilter去重》，作者【九茶】还有另一篇文章可以参考《scrapy_redis去重优化，已有7亿条数据》
去重需要考虑两个点：去重的数据量、去重速度。
为了保持较快的去重速度，一般选择在内存中进行去重。
- 数据量不大时，可以直接放在内存里面进行去重，例如python可以使用set()进行去重。
当去重数据需要持久化时可以使用redis的set数据结构。当数据量再大一点时，可以用不同的加密算法先将长字符串压缩成 16/32/40 个字符，再使用上面两种方法去重；当数据量达到亿（甚至十亿、百亿）数量级时，内存有限，必须用“位”来去重，才能够满足需求。Bloomfilter就是将去重对象映射到几个内存“位”，通过几个位的 0/1值来判断一个对象是否已经存在。然而Bloomfilter运行在一台机器的内存上，不方便持久化（机器down掉就什么都没啦），也不方便分布式爬虫的统一去重。如果可以在Redis上申请内存进行Bloomfilter，以上两个问题就都能解决了。
Bloomfilter算法如何使用位去重，这个百度上有很多解释。简单点说就是有几个seeds，现在申请一段内存空间，一个seed可以和字符串哈希映射到这段内存上的一个位，几个位都为1即表示该字符串已经存在。插入的时候也是，将映射出的几个位都置为1。
需要提醒一下的是Bloomfilter算法会有漏失概率，即不存在的字符串有一定概率被误判为已经存在。这个概率的大小与seeds的数量、申请的内存大小、去重对象的数量有关。下面有一张表，m表示内存大小（多少个位），n表示去重对象的数量，k表示seed的个数。例如我代码中申请了256M，即1<<31（m=2^31，约21.5亿），seed设置了7个。看k=7那一列，当漏失率为8.56e-05时，m/n值为23。所以n = 21.5/23 = 0.93(亿），表示漏失概率为8.56e-05时，256M内存可满足0.93亿条字符串的去重。同理当漏失率为0.000112时，256M内存可满足0.98亿条字符串的去重。
基于Redis的Bloomfilter去重，其实就是利用了Redis的String数据结构，但Redis一个String最大只能512M，所以如果去重的数据量大，需要申请多个去重块（代码中blockNum即表示去重块的数量）。
代码中使用了MD5加密压缩，将字符串压缩到了32个字符（也可用hashlib.sha1()压缩成40个字符）。它有两个作用，一是Bloomfilter对一个很长的字符串哈希映射的时候会出错，经常误判为已存在，压缩后就不再有这个问题；二是压缩后的字符为 0~f 共16中可能，我截取了前两个字符，再根据blockNum将字符串指定到不同的去重块进行去重
总结：基于Redis的Bloomfilter去重，既用上了Bloomfilter的海量去重能力，又用上了Redis的可持久化能力，基于Redis也方便分布式机器的去重。在使用的过程中，要预算好待去重的数据量，则根据上面的表，适当地调整seed的数量和blockNum数量（seed越少肯定去重速度越快，但漏失率越大）。
编写代码
安装依赖
根据github上的资源《BloomFilter_imooc》以及思路来编写bloomfilter的代码。
先前说过，bloom是一种算法，而不是插件也不是软件，它依赖于mmh3，所以需要在虚拟环境中安装mmh3.
然而当我在本机的anaconda虚拟环境内安装时，出现了报错：
g++: error trying to exec 'cc1plus': execvp: 没有那个文件或目录
网上查阅了很多文章，找到一个适合我的：传送，大致原因是电脑上的gcc版本与g++版本不一致引起的。可以打开终端用命令：
gcc -v
g++ -v
来查看两个东西的版本，最终发现用g++的时候报错，于是我安装它：
sudo apt-get install g++
如果是在阿里云服务器，命令改成：
yum install gcc-c++
安装成功后，再次到anaconda虚拟环境中安装mmh3，才成功安装。
编写bloom代码
根据文章《将bloomfilter(布隆过滤器)集成到scrapy-redis中》的指引，作者是将github代码下载到本地目录。
而我为了省事，我在site-package里面写。
在site-package下新建bloofilter_scrapy_redis的package包（带init那种），然后在里面新建文件bloomfilter.py，编写代码：
# -*- coding: utf-8 -*-
# 18-1-21 下午2:22
# RanboSpider
import mmh3
import redis
import math
import time
class PyBloomFilter():
#内置100个随机种子，种子越多需要的内存就越大，内存小的服务器用30个种子就行了
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'):
self.m = math.ceil(capacity*math.log2(math.e)*math.log2(1/error_rate)) #需要的总bit位数
self.k = math.ceil(math.log1p(2)*self.m/capacity) #需要最少的hash次数
self.mem = math.ceil(self.m/8/1024/1024) #需要的多少M内存
self.blocknum = math.ceil(self.mem/512) #需要多少个512M的内存块,value的第一个字符必须是ascii码，所有最多有256个内存块
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)
这里的pool和conn都是单独连接的，实际上在分布式爬虫中是比较不友好的，多台机器的配置就会烦人，这里暂且这样，后期我再改。
是否配置密码
至于是否配置密码，如何配置密码，在bloomfilter.py文件中，有一句：
pool = redis.ConnectionPool(host='127.0.0.1', port=6379, db=0)
conn = redis.StrictRedis(connection_pool=pool)
其中redis.StrictRedis方法，跟踪(ctrl+左键点击)进去，可以看到init初始化方法里面有个password=None
def __init__(self, host='localhost', port=6379,
db=0, password=None, socket_timeout=None,
socket_connect_timeout=None,
socket_keepalive=None, socket_keepalive_options=None,
connection_pool=None, unix_socket_path=None,
encoding='utf-8', encoding_errors='strict',
charset=None, errors=None,
decode_responses=False, retry_on_timeout=False,
ssl=False, ssl_keyfile=None, ssl_certfile=None,
ssl_cert_reqs=None, ssl_ca_certs=None,
max_connections=None):
这里应该是设置password，也就是将服务器redis的权限密码auth设置进来。
pool = redis.ConnectionPool(host='47.98.110.67', port=6379, db=0, password='quinns')
conn = redis.StrictRedis(connection_pool=pool)
即可完成密码的设置。
集成到scrapy_redis中
上面的布隆过滤器代码写好后，需要集成到scrapy_redis中。完成去重任务的是dupefilter.py文件，就要对它进行改造，路径是site-package/scrapy_redis/目录内：
现将刚才编写的布隆选择器导入此文件
from bloomfilter_scrapy_redis.bloomfilter import conn,PyBloomFilter # 从源码包导入布隆
然后在init方法中初始化布隆选择器（这里贴上整个init代码）：
def __init__(self, server, key, debug=False):
"""Initialize the duplicates filter.
Parameters
----------
server : redis.StrictRedis
The redis server instance.
key : str
Redis key Where to store fingerprints.
debug : bool, optional
Whether to log filtered requests.
"""
self.server = server
self.key = key
self.debug = debug
self.logdupes = True
""" 集成布隆过滤器,通过连接池连接redis """
self.bf = PyBloomFilter(conn=conn, key=key)
接下来改动request_seen方法，在里面对request进行判断，如果此次request请求在redis中存在，则直接返回，如果不存在则添加到redis的队列里面去，让爬虫去爬：
def request_seen(self, request):
"""
……
"""
fp = self.request_fingerprint(request)
"""
集成布隆过滤
判断redis是否存在此指纹，如果存在则直接返回true
如果不存在添加指纹到redis，同时返回false
"""
if self.bf.is_exist(fp):
return True
else:
self.bf.add(fp)
return False
""" 集成布隆过滤器，将下方2行代码注释 """
# This returns the number of values added, zero if already exists.
# added = self.server.sadd(self.key, fp)
# return added == 0
到这里即完成了scrapy_redis对布隆过滤器的集成。
测试
在爬虫代码中编写：
# -*- coding: utf-8 -*-
import scrapy
from scrapy_redis.spiders import RedisSpider
from scrapy.http import Request
from urllib import parse
class JobboleSpider(RedisSpider):
name = 'jobbole'
allowd_domains = ["www.gxnhyd.com"]
redis_key = 'jobbole:start_urls'
def parse(self, response):
"""
将当前列表页的每条标的链接拿到 并传给detail进行深入爬取
通过已知列表页码数量 进行循环爬取 就不用翻页了
"""
total = response.css('.item .tl.pl10 a')
for x in total:
title = x.css('::text').extract_first("")
title_url = x.css('::attr(href)').extract_first("")
yield Request(url=parse.urljoin(response.url, title_url), callback=self.parse_detail)
for i in range(1, 10):
next_pages = "http://www.gxnhyd.com/deals/p-%s" % (i)
yield Request(url=next_pages, callback=self.parse)
def parse_detail(self, response):
"""
获取当前详情页的标的信息 包括金额 收益 期限 借款人
投资人列表 - 投资人用户名/投资人投资金额/投资方式/投资时间等
:param response:
:return:
"""
print(response.url)
通过print对爬取情况做观察
开启爬虫后，由于scrapy_redis的特性，需要给redis里面添加start_urls:
lpush jobbole:start_urls http://www.gxnhyd.com/deals [value ...]
爬虫监听到值之后，立即开始爬取，这一步没问题
但是爬完后它空跑了，不会结束，一直空跑。（事实证明，跑空了也不要紧）
二次测试
在第一次测试通过后，我加大了循环次数for i in range(1, 30)，看看是否会出现重复的值，结果报错了。
报错信息与bloom是否重复无关，原因是我之前看到空跑，就主动停止了代码，导致redis报错：
MISCONF Redis is configured to save RDB snapshots, but is currently not able to persist
解决办法在这里《redis异常解决：MISCONF Redis 》，在redis-cli用命令解决这个权限问题：
config set stop-writes-on-bgsave-error no
二次测试后，发现可以正常运行了。然后观察到bloom也生效了，但是还是有空跑的问题
解决空跑（这个办法其实不太好，不推荐）
空跑就是爬虫在爬取完所有的队列有，不会自动停止,而是一直请求请求，然后观察redis-server窗口有memory的提示一直在进行。
解决这个空跑问题参考了一些资料《scrapy-redis所有request爬取完毕，如何解决爬虫空跑问题？ 》
根据scrapy-redis分布式爬虫的原理，多台爬虫主机共享一个爬取队列。当爬取队列中存在request时，爬虫就会取出request进行爬取，如果爬取队列中不存在request时，爬虫就会处于等待状态.
可是，如果所有的request都已经爬取完毕了呢？这件事爬虫程序是不知道的，它无法区分结束和空窗期状态的不同，所以会一直处于上面的那种等待状态，也就是我们说的空跑。
那有没有办法让爬虫区分这种情况，自动结束呢？
- 从背景介绍来看，基于scrapy-redis分布式爬虫的原理，爬虫结束是一个很模糊的概念，在爬虫爬取过程中，爬取队列是一个不断动态变化的过程，随着request的爬取，又会有新的request进入爬取队列。进进出出。
- 爬取速度高于填充速度，就会有队列空窗期（爬取队列中，某一段时间会出现没有request的情况），爬取速度低于填充速度，就不会出现空窗期。所以对于爬虫结束这件事来说，只能模糊定义，没有一个精确的标准。
可以通过限定爬虫自动关闭时间来完成这个任务，在settings配置：
# 爬虫运行超过23.5小时，如果爬虫还没有结束，则自动关闭
CLOSESPIDER_TIMEOUT = 84600
特别注意 ：如果爬虫在规定时限没有把request全部爬取完毕，此时强行停止的话，爬取队列中就还会存有部分request请求。那么爬虫下次开始爬取时，一定要记得在master端对爬取队列进行清空操作。
想象一下，爬虫已经结束的特征是什么？
那就是爬取队列已空，从爬取队列中无法取到request信息。那着手点应该就在从爬取队列中获取request和调度这个部分。查看scrapy-redis源码，我们发现了两个着手点，调度器site-packages\scrapy_redis\schedluer.py和site-packages\scrapy_redis\spiders.py爬虫。
但是爬虫在爬取过程中，队列随时都可能出现暂时的空窗期。想判断爬取队列为空，一般是设定一个时限，如果在一个时段内，队列一直持续为空，那我们可以基本认定这个爬虫已经结束了。
我选择更改调度器，site-packages\scrapy_redis\schedluer.py所以有了如下的改动：
首先在init里面设定一个初始次数
import datetime
def __init__(self, server,
……
……
"""
""" 为解决空跑问题：设定倒计次数 下方根据次数决定何时关闭爬虫，避免空跑"""
self.lostGetRequest = 0
if idle_before_close < 0:
……
……
完整的init方法代码为：
def __init__(self, server,
persist=False,
flush_on_start=False,
queue_key=defaults.SCHEDULER_QUEUE_KEY,
queue_cls=defaults.SCHEDULER_QUEUE_CLASS,
dupefilter_key=defaults.SCHEDULER_DUPEFILTER_KEY,
dupefilter_cls=defaults.SCHEDULER_DUPEFILTER_CLASS,
idle_before_close=0,
serializer=None):
""" 为解决空跑问题：设定倒计次数 下方根据次数决定何时关闭爬虫，避免空跑"""
self.lostGetRequest = 0
if idle_before_close < 0:
raise TypeError("idle_before_close cannot be negative")
self.server = server
self.persist = persist
self.flush_on_start = flush_on_start
self.queue_key = queue_key
self.queue_cls = queue_cls
self.dupefilter_cls = dupefilter_cls
self.dupefilter_key = dupefilter_key
self.idle_before_close = idle_before_close
self.serializer = serializer
self.stats = None
然后到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:
""" 解决空跑问题，这里判断如果获取到request则重置倒计时lostGetRequest """
self.lostGetRequest = 0
self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)
if request is None:
"""
scrapy_reids跑完数据后不会自动停止，会产生空跑情况，一直空跑
每次调度Schedule时如果队列没有数据 则倒计时+1
50次空跑大约费时5分钟,根据项目需求设定次数，满足空跑次数则主动停止并填写停止原因
"""
self.lostGetRequest += 1
if self.lostGetRequest > 10:
self.spider.crawler.engine.close_spider(self.spider, 'Queue is empty,So active end')
return request
这样就可以解决空跑的问题了。（事实证明，高兴得太早）
真正解决空跑（这个也不好，不建议。因为scrapy_redis已处理空跑问题（我也不确定））
真是太年轻，不懂事，我以为按照别人的想法实施，就可以解决空跑的问题了。然后当自己亲自测试的时候，发现并不是那么回事。
scrapy是异步的，而且request队列确实会有空闲状态，如果有空闲状态就会+1，用数字进行累加的话，虽然上编写了重置为0的操作，但貌似是不行的，测试没有那么细致，反正当空闲状态达到N次(关闭条件)的时候，就会自动关闭（request队列还在抽取，也会被关闭），那这就是个bug。
首先
思路是对的，然而用+1的方式出错了。我换了个思路，用时间差来决定是否关闭爬虫。逻辑：
- 时间差是不会存在累加的情况，所以不会有刚才的bug
- 先初始化一个起始时间
- 在每次请求队列的时候刷新起始时间
- 在每次队列为空的时候开始计时
- 计算时间差，如果队列为空的时间减去起始时间的秒数结果大于设定值，则判定为空跑，关闭爬虫
优点
- 通过时间差来判断空跑，解决了刚才的bug；
- 可以根据时间来关闭爬虫，而不是次数，这样对于日后爬虫的监控更精准
具体的代码如下：
现在init方法设定起始时间
为解决空跑问题：设定起始时间
下方根据记录空跑时间end_times与起始时间的时间差来决定何时关闭爬虫，避免空跑
"""
self.strat_times = datetime.datetime.now()
然后到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:
""" 解决空跑问题，这里判断如果获取到request则重置起始时间strat_times """
self.strat_times = datetime.datetime.now()
self.stats.inc_value('scheduler/dequeued/redis', spider=self.spider)
if request is None:
"""
scrapy_reids跑完数据后不会自动停止，会产生空跑情况，一直空跑
每次调度Schedule时如果队列没有数据 则计算end_times
当end_times与start_times的时间差close_times超过N秒，就判定为空跑且进行关闭爬虫的操作
"""
self.end_times = datetime.datetime.now()
self.close_times = (self.end_times - self.strat_times).seconds
print("tihs close_times is : ")
print(self.close_times)
if self.close_times > 180:
self.spider.crawler.engine.close_spider(self.spider, 'Queue is empty,So active end')
return request
看到下图，跑完数据后会根据时间差关闭爬虫
这样才是真正的解决了空跑的问题
最后运行，可以正常关闭爬虫了。但是结束的时候还会有报错信息：
builtins.AttributeError: 'NoneType' object has no attribute 'start_requests'
2017-12-14 16:18:56 [twisted] CRITICAL: Unhandled Error
Traceback (most recent call last):
File "E:\Miniconda\lib\site-packages\scrapy\commands\runspider.py", line 89, in run
self.crawler_process.start()
File "E:\Miniconda\lib\site-packages\scrapy\crawler.py", line 285, in start
reactor.run(installSignalHandlers=False) # blocking call
File "E:\Miniconda\lib\site-packages\twisted\internet\base.py", line 1243, in run
self.mainLoop()
File "E:\Miniconda\lib\site-packages\twisted\internet\base.py", line 1252, in mainLoop
self.runUntilCurrent()
--- <exception caught here> ---
File "E:\Miniconda\lib\site-packages\twisted\internet\base.py", line 878, in runUntilCurrent
call.func(*call.args, **call.kw)
File "E:\Miniconda\lib\site-packages\scrapy\utils\reactor.py", line 41, in __call__
return self._func(*self._a, **self._kw)
File "E:\Miniconda\lib\site-packages\scrapy\core\engine.py", line 137, in _next_request
if self.spider_is_idle(spider) and slot.close_if_idle:
File "E:\Miniconda\lib\site-packages\scrapy\core\engine.py", line 189, in spider_is_idle
if self.slot.start_requests is not None:
builtins.AttributeError: 'NoneType' object has no attribute 'start_requests'
当通过engine.close_spider(spider, ‘reason’)来关闭spider时，有时会出现几个错误之后才能关闭。可能是因为scrapy会开启多个线程同时抓取，然后其中一个线程关闭了spider，其他线程就找不到spider才会报错。
注意事项
编写代码的schedule.py有个next_request方法有这么一句代码：
request = self.queue.pop(block_pop_timeout)
打开同目录的queue.py文件
所以，PriorityQueue和另外两种队列FifoQueue，LifoQueue有所不同，特别需要注意。
如果会使用到timeout这个参数，那么在setting中就只能指定爬取队列为FifoQueue或LifoQueue
# 指定排序爬取地址时使用的队列，
# 默认的 按优先级排序(Scrapy默认)，由sorted set实现的一种非FIFO、LIFO方式。
# 'SCHEDULER_QUEUE_CLASS': 'scrapy_redis.queue.SpiderPriorityQueue',
# 可选的 按先进先出排序（FIFO）
'SCHEDULER_QUEUE_CLASS': 'scrapy_redis.queue.SpiderQueue',
# 可选的 按后进先出排序（LIFO）
# SCHEDULER_QUEUE_CLASS = 'scrapy_redis.queue.SpiderStack'
数据入库测试
经过多次 的mysql入库测试，发现bloomfilter是生效的，而且增量开始之前，对于那么重复的数据对比过滤是非常快的（仅用了500条数据测试），正常爬取500条数据大约1分钟多一点。在爬取过500多数据后，bloomfilter的略过只用了几秒钟，很短的时间。
这个还是很强的，我很高兴
