SQLite → MySQL 数据迁移踩坑实录

---

一、背景与目标

1. 业务场景

前阵子为了回测一个策略，临时将数据都下载到了 SQLite中，主要是方便快捷，但是随着要求的不断增加，还是感觉到诸多不便，为了后续更好的策略分析，准备将数据迁移到 MySQL中，本文仅针对测试数据将存量股票数据从 SQLite 迁移至 MySQL，解决 SQLite 主键约束失效、数据重复、查询性能差的问题，为后续业务提供稳定可靠的结构化数据源。

2. 核心目标

• ✅ 保证数据完整性：无丢失、无冗余
• ✅ 保证数据一致性：SQLite 清洗后与 MySQL 行数完全对齐
• ✅ 保证迁移可重复性：支持多次执行不破坏数据
• ✅ 输出可复用的迁移脚本与文档，方便他人参考

---

二、迁移方案设计

1. 技术选型

• 源库：SQLite（存量脏数据，无严格主键约束）
• 目标库：MySQL（支持严格主键、事务安全）
• 开发语言：Python（pandas 处理数据、pymysql 操作 MySQL）
• 核心策略：
  1. 先清洗 SQLite 重复数据（按 code+date 去重）
  2. 分批插入 MySQL，避免单批次数据过大导致超时
  3. 采用 REPLACE INTO / INSERT 保证幂等性
  4. 最终校验脚本验证全表一致性

2. 关键 SQL 清洗逻辑

-- SQLite 去重获取有效数据（核心！）
SELECT * FROM daily_k GROUP BY code, date;

---

三、踩坑实录与解决方案

🚩 坑1：SQLite 主键形同虚设，重复数据泛滥

• 现象：SQLite 总数据 20774 行，MySQL 迁移后仅 15579 行，行数对不上
• 根因：SQLite 未真正 enforce 主键约束，允许同一 code+date 重复插入多条数据
• 解决方案：
  1. 编写重复数据检测脚本，统计重复行数
  2. 迁移时通过 GROUP BY code, date 去重，只保留唯一有效数据
  3. 最终确认：有效数据为 15579 行，与 MySQL 迁移后行数完全对齐

---

🚩 坑2：REPLACE INTO 导致“假成功”日志

• 现象：迁移脚本日志显示写入 20774 行，但 MySQL 实际行数不变
• 根因：REPLACE INTO 遇到主键冲突时会覆盖旧数据，不会新增缺失行，无法补全历史数据
• 解决方案：
  1. 先 TRUNCATE TABLE daily_k 清空目标表（仅首次迁移）
  2. 改用纯 INSERT 插入去重后数据，保证所有有效行都写入
  3. 迁移脚本改为“先清空、再插入”模式，确保数据完整

---

🚩 坑3：SQL 语法错误（中文符号/表情混入代码）

• 现象：执行迁移脚本时报 near "✅": syntax error
• 根因：手滑将中文注释、表情符号写入 SQL 语句中，导致 SQLite 语法解析失败
• 解决方案：
  1. 迁移脚本中 SQL 部分纯英文，不添加任何中文注释、表情
  2. 注释统一放在 Python 代码中，避免污染 SQL 语句
  3. 上线前先在测试环境验证 SQL 语法正确性

---

🚩 坑4：数字字段空字符串导致类型错误

• 现象：迁移时报 Incorrect integer value: '' for column 'volume'
• 根因：SQLite 中 volume/amount 等数字字段存在空字符串 ''，MySQL 不允许插入空字符串到整数/浮点列
• 解决方案：
  1. 数据清洗时将空字符串 '' 替换为 0
  2. 用 pd.to_numeric 强制转换为数字类型，填充 NaN 为 0
  3. 保证所有数字字段在插入 MySQL 前为合法数值

---

🚩 坑5：校验脚本配置混乱，多次读取不同配置文件

• 现象：校验脚本有时读 config.py，有时写死配置，导致结果不一致
• 根因：迁移与校验脚本未统一配置来源，代码冗余且易出错
• 解决方案：
  1. 新建独立配置文件 db_config_migrate.py，仅用于迁移场景
  2. 所有脚本统一从该配置文件读取 SQLite/MySQL 连接信息
  3. 避免与项目其他配置文件冲突，保证迁移流程隔离

---

四、核心脚本交付

1. 独立配置文件 db_config_migrate.py

# 仅用于迁移场景，与项目其他配置隔离
SQLITE_DB_PATH = "a_stock_final.db"

MYSQL_CONFIG = {
    "host": "localhost",
    "user": "root",
    "password": "你的MySQL密码",
    "database": "a_stock",
    "charset": "utf8mb4"
}

2. 全表迁移脚本 migrate_final_all_in_one.py

import sqlite3
import pymysql
import pandas as pd
from db_config_migrate import MYSQL_CONFIG, SQLITE_DB_PATH

BATCH_SIZE = 2000

ALL_TABLE_MAPPING = {
    "stock_list": {
        "old": "code,code_name,tradeStatus",
        "new": "code,code_name,trade_status"
    },
    "stock_basic": {
        "old": "code,code_name,ipoDate,outDate,type,status",
        "new": "code,code_name,ipo_date,out_date,type,status"
    },
    "daily_k": {
        "old": "date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,tradestatus,pctChg,peTTM,pbMRQ,psTTM,pcfNcfTTM,isST",
        "new": "trade_date,code,open,high,low,close,preclose,volume,amount,adjustflag,turn,trade_status,pct_chg,pe_ttm,pb_mrq,ps_ttm,pcf_ncf_ttm,is_st"
    },
    "weekly_k": {
        "old": "date,code,open,high,low,close,volume,amount,adjustflag,turn,pctChg",
        "new": "trade_date,code,open,high,low,close,volume,amount,adjustflag,turn,pct_chg"
    },
    "monthly_k": {
        "old": "date,code,open,high,low,close,volume,amount,adjustflag,turn,pctChg",
        "new": "trade_date,code,open,high,low,close,volume,amount,adjustflag,turn,pct_chg"
    },
    "stock_industry": {
        "old": "code,code_name,industry,industryClassification,updateDate",
        "new": "code,code_name,industry,industry_classification,update_date"
    },
    "market_daily": {
        "old": "date,up_cnt,down_cnt,limit_up,limit_down,sh_open,sh_pct,sz_open,sz_pct,vol",
        "new": "trade_date,up_cnt,down_cnt,limit_up,limit_down,sh_open,sh_pct,sz_open,sz_pct,vol"
    }
}

NUM_FIELDS = {
    "volume", "amount", "turn", "preclose",
    "pct_chg", "pe_ttm", "pb_mrq", "ps_ttm", "pcf_ncf_ttm",
    "up_cnt", "down_cnt", "limit_up", "limit_down", "vol"
}

def clean_data(df):
    for col in df.columns:
        if col in NUM_FIELDS:
            df[col] = df[col].replace("", 0)
            df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0)
    return df

def migrate_table(table, conn_lite, cursor):
    print(f"\n=== 迁移 {table} ===")
    cfg = ALL_TABLE_MAPPING[table]
    old_cols = cfg["old"]
    new_cols = cfg["new"]

    if table == "daily_k":
        sql_read = f"SELECT {old_cols} FROM {table} GROUP BY code, date"
    else:
        sql_read = f"SELECT {old_cols} FROM {table}"
    
    df = pd.read_sql(sql_read, conn_lite)
    df.columns = new_cols.split(",")
    df = clean_data(df)

    total = len(df)
    print(f"读取总数: {total}")

    placeholders = ",".join(["%s"] * len(df.columns))
    sql = f"REPLACE INTO {table} ({new_cols}) VALUES ({placeholders})"

    for i in range(0, total, BATCH_SIZE):
        batch = df.iloc[i:i+BATCH_SIZE].values.tolist()
        cursor.executemany(sql, batch)
        cursor.connection.commit()
        print(f"  完成批次: {i} ~ {i+len(batch)}")

    print(f"✅ {table} 同步完成！")

def migrate_all():
    conn_lite = sqlite3.connect(SQLITE_DB_PATH)
    conn_mysql = pymysql.connect(**MYSQL_CONFIG)
    cursor = conn_mysql.cursor()

    cursor.execute("SET FOREIGN_KEY_CHECKS=0")
    cursor.execute("SET UNIQUE_CHECKS=0")

    for table in ALL_TABLE_MAPPING:
        migrate_table(table, conn_lite, cursor)

    cursor.execute("SET FOREIGN_KEY_CHECKS=1")
    cursor.execute("SET UNIQUE_CHECKS=1")

    conn_lite.close()
    conn_mysql.close()
    print("\n🎉 所有表同步完成！")

if __name__ == "__main__":
    migrate_all()

3. 最终验收报告脚本 final_verify_report.py

import sqlite3
import pymysql
from db_config_migrate import MYSQL_CONFIG, SQLITE_DB_PATH

TABLES = {
    "stock_list": "股票列表",
    "stock_basic": "股票基本信息",
    "daily_k": "日K线数据(去重后)",
    "weekly_k": "周K线数据",
    "monthly_k": "月K线数据",
    "stock_industry": "股票行业",
    "market_daily": "市场每日统计"
}

def count_sqlite(table):
    conn = sqlite3.connect(SQLITE_DB_PATH)
    cur = conn.cursor()
    if table == "daily_k":
        cur.execute("""
            SELECT COUNT(*) FROM (
                SELECT 1 FROM daily_k GROUP BY code, date
            )
        """)
    else:
        cur.execute(f"SELECT COUNT(*) FROM {table}")
    res = cur.fetchone()[0]
    conn.close()
    return res

def count_mysql(table):
    conn = pymysql.connect(**MYSQL_CONFIG)
    cur = conn.cursor()
    cur.execute(f"SELECT COUNT(*) FROM {table}")
    res = cur.fetchone()[0]
    conn.close()
    return res

def main():
    print("=" * 60)
    print("📊 数据迁移最终验收报告")
    print("=" * 60)

    all_pass = True
    for table, name in TABLES.items():
        s = count_sqlite(table)
        m = count_mysql(table)
        ok = "✅ 一致" if s == m else "❌ 不一致"
        if s != m:
            all_pass = False
        print(f"{name:<12} | {table:<16} | SQLite:{s:>6} | MySQL:{m:>6} | {ok}")

    print("=" * 60)
    if all_pass:
        print("🎉 最终结论：全部表 100% 一致！迁移圆满成功！")
    else:
        print("⚠️ 结论：存在不一致表")
    print("=" * 60)

if __name__ == "__main__":
    main()

---

五、迁移结果验证

最终验收报告

============================================================
📊 数据迁移最终验收报告
============================================================
股票列表       | stock_list       | SQLite:  5193 | MySQL:  5193 | ✅ 一致
股票基本信息   | stock_basic      | SQLite:  5193 | MySQL:  5193 | ✅ 一致
日K线数据(去重后)| daily_k       | SQLite: 15579 | MySQL: 15579 | ✅ 一致
周K线数据      | weekly_k         | SQLite:     0 | MySQL:     0 | ✅ 一致
月K线数据      | monthly_k        | SQLite:     0 | MySQL:     0 | ✅ 一致
股票行业       | stock_industry   | SQLite:  5101 | MySQL:  5101 | ✅ 一致
市场每日统计   | market_daily     | SQLite:     1 | MySQL:     1 | ✅ 一致
============================================================
🎉 最终结论：全部表 100% 一致！迁移圆满成功！
============================================================

---

六、经验总结

1. 源数据校验优先：迁移前务必检查源数据质量，尤其是主键约束、重复数据
2. 幂等性设计：迁移脚本需支持多次执行，避免单次失败后数据混乱
3. 配置统一管理：避免多套配置文件混用，保证脚本行为一致
4. 分步验证：迁移后必须执行全表校验，确保数据无丢失、无冗余
5. 日志清晰可追溯：迁移脚本输出详细日志，便于排查问题

---