Question

MySQL是一个应用极其广泛的关系型数据库，它是开源免费的，可以支持大型数据库，在个人用户和中小企业中成为技术首选。

使用客户端登录MySQL，创建一个供Scrapy使用的数据库，取名为scrapy_db：

$ mysql -hlocalhost -uliushuo -p12345678
...
mysql> CREATE DATABASE scrapy_db CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';
Query OK, 1 row affected (0.00 sec)
mysql> USE scrapy_db;
Database changed

接下来，创建存储书籍数据的表：

mysql> CREATE TABLE books (
 ->   upc      CHAR(16) NOT NULL PRIMARY KEY,
 ->   name      VARCHAR(256) NOT NULL,
 ->   price    VARCHAR(16) NOT NULL,
 ->   review_rating  INT,
 ->   review_num   INT,
 ->   stock    INT
 -> ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
Query OK, 0 rows affected (0.08 sec)

在Python 2中访问MySQL数据库可以使用第三方库MySQL-Python（即MySQLdb），但是MySQLdb不支持Python 3。在Python 3中，可以使用另一个第三方库mysqlclient作为替代，它是基于MySQL-Python开发的，提供了几乎完全相同的接口。因此，在两个Python版本下，可以使用相同的代码访问MySQL。

Python 2使用pip安装MySQL-python：

sudo pip install MySQL-python

Python 3使用pip安装mysqlclient：

sudo pip install mysqlclient

下面是使用MySQLdb将数据写入MySQL数据库的简单示例，与sqlite3的使用几乎完全相同：

import MySQLdb

#连接数据库, 得到Connection 对象
conn = MySQLdb.connect(host='localhost', db='scrapy_db',
    user='liushuo', passwd='12345678', charset='utf8')

#创建Curosr 对象，用来执行SQL语句
cur = conn.cursor()

#创建数据表
cur.execute('CREATE TABLE person (name VARCHAR(32), age INT, sex char(1)) \
    ENGINE=InnoDB DEFAULT CHARSET=utf8')

#插入一条数据
cur.execute('INSERT INTO person VALUES (%s,%s,%s)', ('刘硕', 34, 'M'))

#保存变更，commit 后数据才被实际写入数据库
conn.commit()

#关闭连接
conn.close()

仿照SQLitePipeline实现MySQLPipeline，代码如下：

import MySQLdb

class MySQLPipeline:
 def open_spider(self, spider):
  db = spider.settings.get('MYSQL_DB_NAME', 'scrapy_default')
  host = spider.settings.get('MYSQL_HOST', 'localhost')
 port = spider.settings.get('MYSQL_PORT', 3306)
 user = spider.settings.get('MYSQL_USER', 'root')
 passwd = spider.settings.get('MYSQL_PASSWORD', 'root')

 self.db_conn = MySQLdb.connect(host=host, port=port, db=db,
         user=user, passwd=passwd, charset='utf8')
 self.db_cur = self.db_conn.cursor()

def close_spider(self, spider):
 self.db_conn.commit()
 self.db_conn.close()

def process_item(self, item, spider):
 self.insert_db(item)

 return item

def insert_db(self, item):
 values = (
  item['upc'],
  item['name'],
  item['price'],
  item['review_rating'],
  item['review_num'],
  item['stock'],
 )

 sql = 'INSERT INTO books VALUES (%s,%s,%s,%s,%s,%s)'
 self.db_cur.execute(sql, values)

上述代码结构与SQLitePipeline完全相同，不再赘述。

在配置文件settings.py中指定我们所要使用的MySQL数据库，并启用MySQLPipeline：

MYSQL_DB_NAME = 'scrapy_db'
MYSQL_HOST = 'localhost'
MYSQL_USER = 'liushuo'
MYSQL_PASSWORD = '12345678'

ITEM_PIPELINES = {
  'toscrape_book.pipelines.MySQLPipeline': 401,
}

运行爬虫，并查看数据库：

结果表明，我们成功地将1000条数据存储到了MySQL数据库。

上述代码中，同样是先执行完全部的插入语句（INSERT INTO），最后一次性调用commit方法提交给数据库。或许在某些情况下，我们的确需要每执行一条插入语句，就立即调用commit方法更新数据库，如爬取过程很长，中途可能被迫中断，这样程序就不能执行到最后的commit。如果在上述代码的insert_db方法中直接添加self.db_conn.commit()，又会使程序执行慢得让人无法忍受。为解决以上难题，下面讲解另一种实现方法。

Scrapy框架自身是使用另一个Python框架Twisted编写的程序，Twisted是一个事件驱动型的异步网络框架，鼓励用户编写异步代码，Twisted中提供了以异步方式多线程访问数据库的模块adbapi，使用该模块可以显著提高程序访问数据库的效率。下面是使用adbapi中的连接池访问MySQL数据库的简单示例：

from twisted.internet import reactor, defer
from twisted.enterprise import adbapi
import threading

dbpool = adbapi.ConnectionPool('MySQLdb', host='localhost', database='scrapy_db',
         user='liushuo', password='liushuo', charset='utf8')
def insert_db(tx, item):
 print('In Thread:', threading.get_ident())
 sql = 'INSERT INTO person VALUES (%s, %s, %s)'
 tx.execute(sql, item)

for i in range(1000):
 item = ('person%s' % i, 25, 'M')
 dbpool.runInteraction(insert_db, item)

reactor.run()

上述代码解释如下：

adbapi.ConnectionPool方法可以创建一个数据库连接池对象，其中包含多个连接对象，每个连接对象在独立的线程中工作。adbapi只是提供了异步访问数据库的编程框架，在其内部依然使用MySQLdb、sqlite3这样的库访问数据库。ConnectionPool方法的第一个参数就是用来指定使用哪个库访问数据库，其他参数在创建连接对象时使用。

dbpool.runInteraction(insert_db, item)以异步方式调用instert_db函数，dbpool会选择连接池中的一个连接对象在独立线程中调用insert_db，其中参数item会被传给insert_db的第二个参数，传给insert_db的第一个参数是一个Transaction对象，其接口与Cursor对象类似，可以调用execute方法执行SQL语句，insert_db执行完后，连接对象会自动调用commit方法。

了解了adbapi的使用后，给出第二个版本的MySQLPipeline，代码如下：

from twisted.enterprise import adbapi

class MySQLAsyncPipeline:
 def open_spider(self, spider):
   db = spider.settings.get('MYSQL_DB_NAME', 'scrapy_default')
   host = spider.settings.get('MYSQL_HOST', 'localhost')
   port = spider.settings.get('MYSQL_PORT', 3306)
   user = spider.settings.get('MYSQL_USER', 'root')
   passwd = spider.settings.get('MYSQL_PASSWORD', 'root')

   self.dbpool = adbapi.ConnectionPool('MySQLdb', host=host, db=db,
            user=user, passwd=passwd, charset='utf8')
def close_spider(self, spider):
 self.dbpool.close()

def process_item(self, item, spider):
 self.dbpool.runInteraction(self.insert_db, item)

 return item

def insert_db(self, tx, item):
 values = (
  item['upc'],
  item['name'],
  item['price'],
  item['review_rating'],
  item['review_num'],
  item['stock'],
 )

 sql = 'INSERT INTO books VALUES (%s,%s,%s,%s,%s,%s)'
 tx.execute(sql, values)

通过前面的讲述，相信大家可以轻松理解上述代码，不再过多解释，该版本比之前的版本在执行效率上有显著提高。