Python Scrapy异步MySQL存储实现详解

异步存储与同步存储的区别

在Python开发中，数据存储主要划分为同步与异步两种模式。同步写入方式速度相对较慢，当爬虫抓取速度较快时，容易因写入瓶颈导致数据保存不完整，出现部分数据未能成功入库的情况。相比之下，异步机制能够将爬虫抓取与数据库写入操作分离执行，两者互不阻塞，从而显著提升写入速度，并有效保障数据的完整性。

异步存储实现步骤

实现异步存储数据库主要包含以下关键步骤：

1. 在settings文件中配置MySQL连接所需参数，包括主机地址、用户账号、密码、目标表名及编码格式等。 2. 自定义Pipeline类，并实现from_settings函数。 3. 从twisted.enterprise模块导入adbapi以引入连接池功能。 4. 从pymysql模块导入cursors以引入游标功能。 5. 在from_settings函数中准备数据库连接参数，创建db_pool连接池实例，并返回当前类的对象，同时传入db_pool。 6. 实现初始化函数__init__，在内部将db_pool赋值给self属性。 7. 实现process_item函数： 7.1 调用self.db_pool.runInteraction传入执行插入数据的函数对象及参数，并接收返回结果。 7.2 为query添加addErrback错误回调函数（需参数），用于处理SQL执行失败的情况，并在回调中对错误数据做进一步处理。 8. 实现插入数据的辅助函数，准备SQL语句并执行。 9. 实现错误回调函数，在回调逻辑中对错误数据进行处理。

天堂图片网异步存储实战

接下来，以天堂图片网为例，简要演示异步存储的具体应用流程：

1. 在存储数据前，可手动创建数据库（自定义表名、字段名及字段类型），或通过代码自动创建。

2. 存储数据前，必须先获取到待存储的数据。

import scrapy
from ..items import ImgItem
class IvskySpider(scrapy.Spider):
  name = 'ivsky'
  allowed_domains = ['ivsky.com']
  start_urls = ['http://www.ivsky.com/tupian/ziranfengguang/']
  def parse(self, response):
    imgs = response.xpath('//div[@class="il_img"]/a/img')
    for img in imgs:
      alt = img.xpath('@alt').extract_first('')
      src = img.xpath('@src').extract_first('')
      item = ImgItem()
      item['alt'] = alt
      item['src'] = src
 
      yield item

3. 自定义Item类，并将获取到的数据传入其中。

import scrapy
 
class IvskySpiderItem(scrapy.Item):
  # define the fields for your item here like:
  # name = scrapy.Field()
  pass
 
class ImgItem(scrapy.Item):
 
  alt = scrapy.Field()
  src = scrapy.Field()

4. 接下来进行settings配置，相关代码如下（注意将robots协议设置为False）：

MYSQL_HOST = '127.0.0.1'
MYSQL_USER = 'root'
MYSQL_PW = '123456'
MYSQL_DB = 'ivskydb'
MYSQL_CHARSET = 'utf8'

5. 随后自定义Pipeline，并在settings中对该Pipeline进行配置（需设置优先级）：

from twisted.enterprise import adbapi
from pymysql import cursors
 
class TwistedMysqlPipeline(object):
 
  # 在调用TwistedMysqlPipeline时，第一个调用该函数
  @classmethod
  def from_settings(cls, settings):
 
    #准备需要用到的链接mysql的参数
    db_prams = dict(
      host=settings['MYSQL_HOST'],
      user=settings['MYSQL_USER'],
      password=settings['MYSQL_PW'],
      db=settings['MYSQL_DB'],
      port=3306,
      use_unicode=True,
      charset=settings['MYSQL_CHARSET'],
      # 指定使用的游标类型
      cursorclass=cursors.DictCursor
    )
    # 创建连接池对象，需要传入两个参数
    # 1.使用操作mysql第三方包名
    # 2.连接数据库需要的参数
    db_pool = adbapi.ConnectionPool('pymysql', **db_prams)
 
    return cls(db_pool)
 
  def __init__(self, db_pool):
    # 将连接池对象赋值self.db_pool属性
    self.db_pool = db_pool
 
  def process_item(self, item, spider):
 
    # 准备sql
    # 执行sql
    # 执行一个将item数据写入数据库的动作
    # 1.执行操作的函数
    # 2.执行函数需要的参数....
    query = self.db_pool.runInteraction(self.insert_item, item)
    # 执行sql出现异常错误时，回调的函数
    query.addErrback(self.handle_error, item, spider)
 
    return item
 
  # 插入数据出现错误时，回调的函数
  def handle_error(self, failure, item, spider):
    print(failure)
    print(item)
 
  # 执行插入数据的函数
  def insert_item(self, cursor, item):
    # 创建sql
    sql = "INSERT INTO ivs(alt,src)VALUES(%s,%s)"
    # 执行sql
    cursor.execute(sql,(item['alt'], item['src']))

6. 在settings中完成Pipeline的配置。

ITEM_PIPELINES = {
  # 'ivsky_spider.pipelines.MysqlPipeline': 300,
  'ivsky_spider.pipelines.TwistedMysqlPipeline': 300,
}

代码部分至此结束。

以上就是本文的全部内容，希望对各位的学习有所帮助，也欢迎大家多多支持本站。