非本地MySQL数据库INSERT查询速度极慢问题排查求助

我用VSCode运行Python 3.10脚本，通过SSH隧道往PythonAnywhere上的MySQL 8数据库插入数据。插入275行数据耗时约50秒（对应JSON数据大小750KB，实际数据量略小），过程无报错。但同样的脚本往本地同版本数据库插入完全相同的数据仅需0.2秒。调整提交批次后，总耗时从75秒降到50秒，但这已经是最优效果了。这个脚本是给网络爬虫用的，正式运行要处理数百到数千行数据的增删，现在的性能完全不够。计时器只包裹了INSERT查询，没算SSH连接这些耗时。

请问问题出在哪？

表结构（无额外索引） #

listingID int PRIMARY KEY AUTO_INCREMENT, 
types VARCHAR(11), 
town VARCHAR(255), 
postcode CHAR(5), 
price INT UNSIGNED, 
agent VARCHAR(50), 
ref VARCHAR(30), 
bedrooms SMALLINT UNSIGNED, 
rooms SMALLINT UNSIGNED, 
plot MEDIUMINT UNSIGNED, 
size MEDIUMINT UNSIGNED, 
link_url VARCHAR(1024), 
description VARCHAR(14000), 
photos TEXT, 
photos_hosted TEXT, 
gps POINT, 
id VARCHAR(80), 
types_original VARCHAR(30)

数据插入函数 #

listings是字典列表，原本用mysql.connector的命名占位符，后来换成不支持该特性的MySQLdb，所以用字典处理插入逻辑：

def insert_data_to_table(cursor, table_name, columns_list, values_dict, gps_string):
    # 创建值对应的csv格式%s占位符字符串
    placeholders = ", ".join(f"%s" for _ in values_dict.keys())

    # 创建要插入的列名字符串
    columns = ", ".join(columns_list)

    # 如果有GPS数据，就在查询中加入对应的字符串；否则构建不带GPS的查询
    if gps_string:
        insert_query = f"INSERT INTO {table_name} (gps, {columns}) VALUES ({gps_string}, {placeholders})"
    else:
        insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"

    cursor.execute(insert_query, tuple(values_dict.values()))

def add_listings(cursor, listings):
    print("Adding data...")
    counter = 0
    for listing in listings:
        # 每50行提交一次比每行提交耗时减少约33%，20、50、100行的批次差别不大
        if counter > 50:
            print("Commit", counter)
            db.commit()
            counter = 0
        columns_list = []

        values_dict = {key: None for key in listing if key != "gps"}

        if listing.get("gps") is None:
            gps_string = None
        elif isinstance(listing.get("gps"), list):
            gps_string = f"ST_GeomFromText('POINT({round(listing['gps'][0], 6)} {round(listing['gps'][1], 6)})', 4326)"

        for key in values_dict:
            if isinstance(listing.get(key), list):
                values_dict[key] = ":;:.".join([str(x) for x in listing[key]])
            else:
                values_dict[key] = listing.get(key)
            columns_list.append(key)

        insert_data_to_table(cursor, table_name, columns_list, values_dict, gps_string)
        counter += 1
    db.commit()

问题原因与优化方案 #

核心原因：单条插入的网络往返开销 #

当前代码逐行执行INSERT语句，哪怕每50行提交一次，每次cursor.execute仍会单独向远程数据库发送一次SQL请求。远程网络的往返延迟（RTT）是本地的数百倍，275次请求的往返时间累加，直接导致了50秒的耗时——本地因为网络延迟可以忽略，所以仅需0.2秒。

最有效的优化：批量插入 #

将多行数据合并为单条批量INSERT语句，把网络请求次数从275次压缩到6次左右（按50行批次），直接砍掉大部分网络开销。优化后的批量插入函数示例：

def add_listings_bulk(cursor, listings, table_name, batch_size=50):
    print("Adding data in bulk...")
    batch_values = []
    batch_columns = None
    has_gps = False

    for listing in listings:
        # 整理当前行的非GPS字段
        values_dict = {key: listing.get(key) for key in listing if key != "gps"}
        # 统一处理列表类型字段
        for k, v in values_dict.items():
            if isinstance(v, list):
                values_dict[k] = ":;:.".join([str(x) for x in v])
        
        current_columns = list(values_dict.keys())
        # 初始化批次的列结构（假设所有listings字段一致）
        if batch_columns is None:
            batch_columns = current_columns
            has_gps = listing.get("gps") is not None
        
        # 组装当前行的插入值
        if has_gps and isinstance(listing.get("gps"), list):
            gps_val = f"ST_GeomFromText('POINT({round(listing['gps'][0],6)} {round(listing['gps'][1],6)})',4326)"
            row_values = [gps_val] + list(values_dict.values())
        else:
            row_values = list(values_dict.values())
        
        batch_values.append(tuple(row_values))
        
        # 达到批次大小则执行批量插入
        if len(batch_values) >= batch_size:
            # 构建批量插入SQL
            if has_gps:
                columns_str = ", ".join(["gps"] + batch_columns)
                placeholder_count = len(batch_columns) + 1
                placeholders_str = ", ".join(["%s"] * placeholder_count)
                values_groups = ", ".join([f"({placeholders_str})"] * len(batch_values))
                insert_query = f"INSERT INTO {table_name} ({columns_str}) VALUES {values_groups}"
            else:
                columns_str = ", ".join(batch_columns)
                placeholder_count = len(batch_columns)
                placeholders_str = ", ".join(["%s"] * placeholder_count)
                values_groups = ", ".join([f"({placeholders_str})"] * len(batch_values))
                insert_query = f"INSERT INTO {table_name} ({columns_str}) VALUES {values_groups}"
            
            # 执行插入并提交
            cursor.execute(insert_query, sum(batch_values, ()))
            db.commit()
            print(f"Committed {len(batch_values)} rows")
            batch_values = []
    
    # 处理剩余的不足批次的数据
    if batch_values:
        if has_gps:
            columns_str = ", ".join(["gps"] + batch_columns)
            placeholder_count = len(batch_columns) + 1
            placeholders_str = ", ".join(["%s"] * placeholder_count)
            values_groups = ", ".join([f"({placeholders_str})"] * len(batch_values))
            insert_query = f"INSERT INTO {table_name} ({columns_str}) VALUES {values_groups}"
        else:
            columns_str = ", ".join(batch_columns)
            placeholder_count = len(batch_columns)
            placeholders_str = ", ".join(["%s"] * placeholder_count)
            values_groups = ", ".join([f"({placeholders_str})"] * len(batch_values))
            insert_query = f"INSERT INTO {table_name} ({columns_str}) VALUES {values_groups}"
        
        cursor.execute(insert_query, sum(batch_values, ()))
        db.commit()
        print(f"Committed remaining {len(batch_values)} rows")

注意：如果listings中每行字段不统一，需先统一列结构再批量插入；GPS字段使用SQL函数时，要确保数据为可信来源，避免SQL注入风险。

其他辅助优化 #

1. SSH隧道优化：开启SSH压缩（添加-C参数），减少传输数据量；检查PythonAnywhere网络出口是否有带宽或延迟波动。
2. 字段类型适配：确认TEXT类型的photos、photos_hosted字段数据大小，批量插入能更高效利用带宽。
3. 数据库资源检查：确认PythonAnywhere的MySQL实例CPU、IO资源是否被其他任务占用，导致插入变慢。

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