Scrapy笔记：持久化，Feed exports的使用

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

首先要明确的是，其实所有的FeedExporter都是类，里面封装了一般进行io操作的方法。因此，要怎么输出呢？其实从技术实现来说，在生成item的每一步调用其进行储存都是可以的，只不过为了更加符合scrapy的架构，一般都是在Pipeline中使用FeedExporter的。

每一个Exporter的使用都是类似的：

在settings.py中写入相应的配置，

在pipeline中调用exporter：

exporter.start_exporter()

exporter.export_item()

exporter.finish_exporter()

其它工作都已经由scrapy封装好了，所以就不需要再进行额外设定了。

由于item的输出一般是连续输出的，因此可以将export开始和结束的方法放到spider_opened和spider_closed中启动。

以将item输出到json文件为例，下面是相关的配置和写法：

在settings.py中的配置：

FEED_FORMAT = 'json' # 输出格式
FEED_EXPORTERS_BASE = {
'json': 'scrapy.exporters.JsonItemExporter',
'jsonlines': 'scrapy.exporters.JsonLinesItemExporter',
}

在pipeline中的设定：

class MyCustomPipeline(object):
def __init__(self):
self.files = {} @classmethod
def from_crawler(cls, crawler): # 生成pipeline实例的方法
pipeline = cls()
crawler.signals.connect(pipeline.spider_opened, signals.spider_opened) # 将spider_opened连接到信号上，当spider打开时执行spider_opened方法
crawler.signals.connect(pipeline.spider_closed, signals.spider_closed)
return pipeline
def spider_opened(self, spider): #
file = open('%s_ip.json' % spider.name, 'w+b') # 生成文件描述符
self.files[spider] = file # 保存描述符的引用
self.exporter = JsonLinesItemExporter(file) # 实例化一个Exporter类
self.exporter.start_exporting() # 开始输出 def spider_closed(self,spider):
self.exporter.finish_exporting() # 结束输出
#print('*'*50)
file = self.files.pop(spider)
#print(file.name)
file.close() def process_item(self, item, spider):
self.exporter.export_item(item) # 正式输出
return item

那么怎样输出到mysql数据库中呢？

实际上scrapy自带的FeedExporter中并没有输出到关系型数据库的输出器，因此只能在pipelines中自己定义函数进行处理。由于scrapy是基于twisted异步框架开发的，使用传统的MySQLdb等mysql连接库会出现阻塞。为此，twisted提供了异步数据库实现方法，也就是使用连接池的方式进行交互。

from twisted.enterprise import adbapi
self.dbpool = adbapi.ConnectPool(xxxx) # 生成连接池对象
yield self.dbpool.runInteraction(interaction_function, arg) # 返回异步处理数据库交互的方法

具体使用：

假设已经在配置文件settings.py中设定了

MYSQL_PIPELINE_URI = 'mysql://root:root@localhost/proxyip' #MySQL的uri

pipelines.py文件中的设置：

class MySQLPipeline(object): def __init__(self, mysql_url):
'''创建连接池'''
# 储存以便将来引用
self.mysql_url = mysql_url
# 报告连接错误
self.report_connection_error = True
# 解析mysql的uri，并初始化dbpool
conn_kwargs = MySQLPipeline.parse_mysql_url(mysql_url)
self.dbpool = adbapi.ConnectionPool('MySQLdb',
charset='utf8',
use_unicode=True,
connect_timeout=5,
**conn_kwargs) @classmethod
def from_crawler(cls, crawler):
'''检索crawler，获取settings'''
# Get url from settings
mysql_url = crawler.settings.get('MYSQL_PIPELINE_URI', None)
# 如果没有配置uri，触发错误
if not mysql_url:
raise NotConfigured
# 生成MySQLPipeline实例
return cls(mysql_url) def close_spider(self, spider):
'''spider关闭时关闭连接池'''
self.dbpool.close()
@defer.inlineCallbacks
def process_item(self, item, spider):
'''处理item，将其传入mysql数据库'''
logger = spider.logger
try:
yield self.dbpool.runInteraction(MySQLPipeline._do_replace, item)
except MySQLdb.OperationalError:
if self.report_connection_error:
print('Can not connect to MySQL:%s'%self.mysql_url)
self.report_connection_error = False else:
print(traceback.format_exc())
# 返回item给下一阶段
defer.returnValue(item) @staticmethod
def _do_replace(tx, item):
'''实现具体的替换操作'''
sql = '''INSERT INTO ips(ip, port, protocol, speed, auth_time, is_transparent) VALUES(%s, %s, %s, %s, %s, %s)'''
args = (
item['ip'],
item['port'],
item['protocol'],
item['speed'],
item['auth_time'],
item['is_transparent'],
)
tx.execute(sql, args) @staticmethod
def parse_mysql_url(mysql_url):
'''通过url获取数据库连接的参数，提供给adbapi的连接池''' params = dj_database_url.parse(mysql_url)
conn_kwargs = {}
conn_kwargs['host'] = params['HOST']
conn_kwargs['user'] = params['USER']
conn_kwargs['passwd'] = params['PASSWORD']
conn_kwargs['db'] = params['NAME']
conn_kwargs['port'] = params['PORT']
# 删除空值
conn_kwargs = dict((k,v) for k,v in conn_kwargs.iteritems() if v) return conn_kwargs