Scrapy批量插入MySQL的SQL优化方案咨询（附现有代码）

问题背景 #

已实现Scrapy缓存1000条数据，但缓存数据仍逐条插入MySQL导致速度缓慢。存储逻辑为：

• 将数据插入products表
• 若product_id不存在于products表，则同时插入new_products表（后台有脚本清理旧数据，总数据量最多5万条）

推测性能瓶颈在于插入new_products时的product_id存在性检查逻辑，现需要实现一次性插入1000条数据的优化方案。

当前pipeline.py代码 #

from __future__ import print_function
import logging
from scrapy import signals
from itemadapter import ItemAdapter
from mysql.connector import errorcode
from amazon_scraper.items import AmazonMobileDetailsItem
import mysql.connector


class AmazonScraperPipeline:
    table = 'products'
    table2 = 'new_products'
    conf = {
        'host': 'localhost',
        'user': 'xxxxxx',
        'password': 'xxxxxx',
        'database': 'xxxxxxx',
        'raise_on_warnings': True
    }
    
    def __init__(self,**kwargs):
        self._rows = []   #  store rows temporarily
        self._cached_rows = 0    # number of cached rows
        self._cache_limit = 1000   # limit before saving to database
        self.cnx = self.mysql_connect()

    def open_spider(self, spider):
        print("spider open")

    def save_all(self):    # calls self.save method for all cached rows
        if len(self._rows) >0:
            list(map(self.save, self._rows))
            self._cached_rows =0   # reset the count
            self._rows = []         # reset the cache

    def cache_result(self, item):  # adds new row to cache
        self._rows.append(dict(item))
        self._cached_rows +=1
        if self._cached_rows >= self._cache_limit: # checks if limit reached
            self.save_all()      # if it has been reached then save all rows

    def process_item(self, item, spider):
        print("Saving item into db ...")
        self.cache_result(item)    # cache this item
        return item

    def close_spider(self, spider):
        self.save_all()      # Saves remaining rows once spider closes
        self.cnx.close()

    def mysql_connect(self):
        try:
            return mysql.connector.connect(**self.conf)
        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                print("Something is wrong with your user name or password")
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                print("Database does not exist")
            else:
                print(err)


    def save(self, row):
        cursor = self.cnx.cursor()
        cursor.execute("SELECT DISTINCT product_id FROM products;")
        existing_ids = [row[0] for row in cursor.fetchall()]
        create_query = ("INSERT INTO " + self.table +
            "(rowid, date, listing_id, product_id, product_name, price, url) "
            "VALUES (%(rowid)s, %(date)s, %(listing_id)s, %(product_id)s, %(product_name)s, %(price)s, %(url)s)")
        # data_user = (rowid, date, listing_id, product_id, product_name, price, url)
        # Insert new row
        cursor.execute(create_query, row)
        # lastRecordId = cursor.lastrowid

        # Make sure data is committed to the database
        # self.cnx.commit()
        # cursor.close()
        print("Item saved")

        product_id = row['product_id']
        if not product_id in existing_ids:
            create_query = ("INSERT INTO " + self.table2 +
                "(product_rowid, date, listing_id, product_id, product_name, price, url) "
                "VALUES (%(rowid)s, %(date)s, %(listing_id)s, %(product_id)s, %(product_name)s, %(price)s, %(url)s)")
            # data_user = (rowid, date, listing_id, product_id, product_name, price, url)

            # new_cursor = self.cnx.cursor()
            cursor.execute(create_query, row)
            # lastRecordId = cursor.lastrowid
            # self.cnx.commit()
            # new_cursor.close()
            print("New Item saved")
        self.cnx.commit()

优化方案 #

核心优化方向 #

• 批量插入代替逐条插入，减少数据库交互次数
• 预查询现有product_id并存储为集合，提升存在性检查效率
• 复用数据库游标，减少资源开销
• 合并事务提交，降低事务处理成本

优化后的pipeline.py代码 #

from __future__ import print_function
import logging
from scrapy import signals
from itemadapter import ItemAdapter
from mysql.connector import errorcode
from amazon_scraper.items import AmazonMobileDetailsItem
import mysql.connector


class AmazonScraperPipeline:
    table = 'products'
    table2 = 'new_products'
    conf = {
        'host': 'localhost',
        'user': 'xxxxxx',
        'password': 'xxxxxx',
        'database': 'xxxxxxx',
        'raise_on_warnings': True
    }
    
    def __init__(self,**kwargs):
        self._rows = []   # 临时存储行数据
        self._cache_limit = 1000   # 批量插入阈值
        self.cnx = self.mysql_connect()
        # 预加载现有product_id，用集合存储提升查询速度
        self.existing_product_ids = self._get_existing_product_ids()

    def open_spider(self, spider):
        print("spider open")

    def _get_existing_product_ids(self):
        """一次性获取products表中所有已存在的product_id"""
        cursor = self.cnx.cursor()
        cursor.execute("SELECT DISTINCT product_id FROM products;")
        # 转换为集合，O(1)时间复杂度的存在性检查
        ids = {row[0] for row in cursor.fetchall()}
        cursor.close()
        return ids

    def save_all(self):    # 批量处理缓存的所有数据
        if not self._rows:
            return
        
        cursor = self.cnx.cursor()
        
        # 1. 批量插入products表
        products_insert_query = f"""
            INSERT INTO {self.table} 
            (rowid, date, listing_id, product_id, product_name, price, url)
            VALUES (%(rowid)s, %(date)s, %(listing_id)s, %(product_id)s, %(product_name)s, %(price)s, %(url)s)
        """
        # 使用executemany执行批量插入，大幅减少数据库交互
        cursor.executemany(products_insert_query, self._rows)
        print(f"批量插入{len(self._rows)}条数据到products表")

        # 2. 筛选需要插入new_products的数据
        new_products_rows = [row for row in self._rows if row['product_id'] not in self.existing_product_ids]
        if new_products_rows:
            new_products_insert_query = f"""
                INSERT INTO {self.table2} 
                (product_rowid, date, listing_id, product_id, product_name, price, url)
                VALUES (%(rowid)s, %(date)s, %(listing_id)s, %(product_id)s, %(product_name)s, %(price)s, %(url)s)
            """
            cursor.executemany(new_products_insert_query, new_products_rows)
            print(f"批量插入{len(new_products_rows)}条数据到new_products表")
            # 更新现有ID集合，确保后续批次检查准确
            self.existing_product_ids.update(row['product_id'] for row in new_products_rows)
        
        # 统一提交事务，减少事务开销
        self.cnx.commit()
        cursor.close()
        
        # 重置缓存
        self._rows = []
        print("批量插入完成")

    def cache_result(self, item):  # 添加新数据到缓存
        self._rows.append(dict(item))
        if len(self._rows) >= self._cache_limit: # 检查是否达到批量阈值
            self.save_all()      # 执行批量插入

    def process_item(self, item, spider):
        print("缓存item到本地 ...")
        self.cache_result(item)    # 缓存当前item
        return item

    def close_spider(self, spider):
        self.save_all()      # 爬虫关闭时保存剩余数据
        self.cnx.close()

    def mysql_connect(self):
        try:
            return mysql.connector.connect(**self.conf)
        except mysql.connector.Error as err:
            if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
                print("用户名或密码错误")
            elif err.errno == errorcode.ER_BAD_DB_ERROR:
                print("数据库不存在")
            else:
                print(err)

关键优化说明 #

1. 预加载现有ID集合：初始化时一次性查询所有product_id并存储为集合，集合的in操作是O(1)，远快于列表的O(n)，且避免了每条数据都执行全表查询的开销。
2. 批量插入：使用executemany方法代替多次execute，将1000条数据的插入请求合并为一次交互，大幅降低网络往返和数据库处理次数。
3. 事务合并：批量操作完成后统一提交事务，减少事务创建和提交的开销。
4. 动态更新ID集合：插入new_products后立即更新现有ID集合，确保后续缓存批次的存在性检查准确，无需重复查询数据库。
5. 游标复用：在批量处理过程中只创建一次游标，避免频繁创建和销毁游标带来的资源消耗。

内容的提问来源于stack exchange，提问作者cevreyolu