1. 背景为什么会有“古早库存”以及它带来的问题看到这个标题很多做后台开发或者数据治理的同学应该会心一笑。“古早库存”翻译过来就是老系统里遗留了很久的历史数据“搬家”放到工程语境里就是系统迁移、数据库搬迁、留了几年的老库终于要换了。这句话看起来像一条日常动态但背后其实是一个非常典型的业务场景旧的库存数据要怎么搬到新系统里搬完之后怎么保证数据是对的。在实际项目中这种“搬家”场景特别常见公司从老 ERP 切换到新 ERP老系统里跑了五六年库存数据积累了几十万条。仓库系统升级原来用的 Excel 表格 老单机程序终于要换成正儿八经的数据库系统。公司搬迁、机房更换、数据库版本升级顺手要把老数据同步到新库。业务并购两边系统合并库存数据要先盘点再搬过去。有人可能会想库存数据不就是一个“数量”字段吗直接复制过去不就行了如果你真这么干大概率会在上线第二天接到仓库主管的电话“这个商品库存怎么变成负数了”“为什么同一个货号有三条记录”“这个单位明明是箱怎么搬过去变成件了”所以这篇文章不讨论“库存数据到底应该存哪个字段”而是把“库存数据迁移”这件事从头到尾拆开讲清楚迁移前要做什么、迁移中怎么写代码、迁移后怎么核对以及最容易踩的坑在哪里。内容偏向实战既有方案设计也有可运行的 Python 迁移脚本示例适合后端开发、数据开发、运维和独立开发者的项目落地参考。细心的读者会发现标题说“搬家之前的库存”其实就是在提醒一件很重要的事搬家之前先要想清楚这些库存数据是哪些、长什么样、有没有用而不是等到搬完才去翻旧账。下面我们以“老库存系统迁到新库存系统”为场景完整走一遍。2. 迁移前准备盘点、评估与目标设计很多迁移项目失败不是因为写不出迁移代码而是因为没搞清楚老数据里到底有什么。所以第一件事不是写 SQL而是做一次数据盘点。2.1 数据盘点清单先整理出需要迁移的数据范围一般包括数据类别必查字段关注点商品主数据商品编码、名称、规格、条码编码是否有重复、是否包含特殊字符库存余额表商品编码、仓库编码、数量、单位、更新时间数量是否有负数、单位是否统一仓库档案仓库编码、仓库名称、状态是否存在已停用仓库历史出入库流水单据号、类型、商品、数量、时间流水的覆盖时间范围、数据量大小组织/部门数据部门编码、名称、层级与新系统组织架构是否对应在盘点阶段要输出一张表说明每一类数据大概多少条、总体积多大、最早一条数据是什么时候、最新一条是什么时候。这个信息后面决定迁移方式。2.2 制定迁移范围和规则老数据不是全部要搬。库存迁移最常见的规则有两种余额迁移把当前最新的库存余额搬过去不搬历史流水。适合仓库系统重构、只保留业务当前状态的场景。全量迁移余额和所有出入库流水一起搬保留历史追溯能力。适合财务审计要求严格的业务。一般来说如果新系统没有太强的历史单据追溯需求建议只迁移余额和必要的商品主数据。全量迁移听着好听但流水数据格式千差万别清洗成本成倍上升而且业务方大概率也不会去翻五年前的入库单。2.3 环境准备与版本说明迁移脚本的运行环境和数据库环境需要提前统一。本文示例以常见环境为例重点演示配置思路具体版本需要根据你的实际项目情况调整操作系统Windows / Linux 均可。编程语言Python 3.8使用pymysql和pandas。数据源MySQL 5.7 / 8.0或其他支持 SQL 的数据库。目标库与数据源同一类型的 MySQL 实例方便演示。工具Navicat / DBeaver / 命令行 mysql 客户端。如果你用的是 Oracle、PostgreSQL 或者 SQL ServerSQL 语法层面的兼容需要另外调整但迁移的思路是通用的。3. 数据清洗把“古早库存”变成能搬的库存老数据之所以“古早”就是因为它不规范。数据清洗是整个迁移过程中最花时间的环节下面列举几个最常见的库存数据问题。3.1 商品编码不统一同一个商品在老系统里可能有多个编码。比如“A001”和“A0001”其实是同一个商品但一个来自手工录入一个来自导入模板。如果不处理迁移后新系统会出现重复商品库存数量被拆成两半对账永远对不上。处理方法先做编码归一化。把所有编码统一去掉多余的前导零然后做一次去重或编码映射。# 编码归一化示例 def normalize_code(code: str) - str: if not code: return code str(code).strip() # 去掉多余前导零但保留纯数字串的原始意义 if code.isdigit(): code str(int(code)) return code.upper()注意如果编码本身包含字母或有业务含义不能直接去掉前导零必须先和业务确认编码规则。3.2 数量单位不统一这是库存迁移里最坑的问题。有的记录单位是“箱”有的单位是“件”还有的字段里直接存了“100箱”这种带单位的文本。迁移时如果不统一单位库存数量直接错一大截。处理策略是在迁移前建立一张单位换算表把老数量先换算成最小库存单位再写入新系统。unit_map { 箱: 12, # 1箱 12件 件: 1, 打: 12, 包: 20, } def convert_to_base_unit(quantity: float, unit: str) - float: unit str(unit).strip() if unit not in unit_map: raise ValueError(f未知单位: {unit}) return quantity * unit_map[unit]3.3 数量为负数或空值库存余额理论上不该有负数但实际老系统经常出现负数库存原因是出库先于入库或者盘点差异没有及时调整。这类数据直接搬过去会导致新系统后续的库存计算直接混乱。清洗策略空值按 0 处理单独登记。负数要单独拉出来由业务确认后再迁移不能自作主张改成 0。def clean_quantity(value): if value is None: return 0, NULL转0 if value 0: return value, 负库存待业务确认 return value, 正常3.4 重复数据由于历史导入脚本重复执行同一条商品 同仓库的库存记录可能出现多次。迁移前要去重。# 按商品编码 仓库编码去重保留更新时间最新的记录 df df.sort_values(update_time, ascendingFalse) df df.drop_duplicates(subset[product_code, warehouse_code], keepfirst)4. 库存数据迁移方案设计清洗完成之后进入迁移方案设计。这个环节决定上线时怎么切也是整个迁移项目中最核心的一步。4.1 方案A全量停机迁移适合业务量不大、允许短时间停机的场景。流程如下业务停止操作。备份老库数据。执行全量迁移脚本把清洗后的数据写入新库。执行对账确认数量一致。切换业务系统到新库。优点逻辑简单容易核对。 缺点停机时间长业务影响范围大。4.2 方案B全量迁移 增量同步适合数据量大、停机时间不能太长的场景。流程如下先跑一次全量迁移。在迁移过程中记录老库业务操作的日志表或时间戳。全量迁移完成后再跑一次增量同步把迁移期间产生的变化同步到新库。最后短暂停机做最终一致性校验后切换。这种方式比停机迁移复杂但对大库存、连续生产业务的场景友好很多。4.3 方案C双写双写是在新系统正式切换前让业务同时写入老库和新库运行一段时间确认新库稳定后再下线老库。双写的问题在于如果两边代码事务不统一很容易造成数据不一致调试成本较高。很多小团队并不适合直接上双写一般建议先用“全量 增量”或者“停机迁移”解决。4.4 迁移工具选型如果数据量不大直接用 Python 脚本 SQL 就够了。如果数据量达到百万级以上建议用成熟的 ETL 工具如 DataX、Kettle或者在数据库层面用INSERT ... SELECT的方式做批量迁移。选择工具的核心原则是可回滚、可重跑、有日志。不管用什么工具迁移脚本必须支持断点续跑不能跑到一半失败就全部从头再来。5. 实战案例Python 实现库存迁移下面用一个最小可运行的示例演示从老库读取库存数据、清洗、写入新库的完整过程。为了便于理解我们使用两张表老库old_db.stock_balanceCREATE TABLE stock_balance ( id int(11) NOT NULL AUTO_INCREMENT, product_code varchar(32) DEFAULT NULL, warehouse_code varchar(32) DEFAULT NULL, quantity decimal(12,2) DEFAULT NULL, unit varchar(16) DEFAULT NULL, update_time datetime DEFAULT NULL, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;新库new_db.stock_balance_newCREATE TABLE stock_balance_new ( id bigint(20) NOT NULL AUTO_INCREMENT, product_code varchar(32) NOT NULL, warehouse_code varchar(32) NOT NULL, quantity decimal(12,2) NOT NULL DEFAULT 0, unit varchar(16) NOT NULL DEFAULT 件, update_time datetime DEFAULT NULL, PRIMARY KEY (id), UNIQUE KEY uk_product_warehouse (product_code, warehouse_code) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;新表增加了一个唯一键避免迁移后同一个商品 仓库出现多条记录。5.1 安装依赖pip install pymysql pandas5.2 编写迁移脚本import pymysql import pandas as pd from datetime import datetime # 数据库连接配置 SOURCE_DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: your_password, database: old_db, charset: utf8mb4, } TARGET_DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, password: your_password, database: new_db, charset: utf8mb4, } UNIT_MAP { 件: 1, 箱: 12, 打: 12, 包: 20, 个: 1, } def normalize_code(code): if code is None: return code str(code).strip() if code.isdigit(): code str(int(code)) return code.upper() def convert_quantity(quantity, unit): if quantity is None: return 0.0, NULL转0 unit str(unit).strip() if unit not in UNIT_MAP: raise ValueError(f未知单位: {unit}) return float(quantity) * UNIT_MAP[unit], 正常 def fetch_source_data(): conn pymysql.connect(**SOURCE_DB_CONFIG) sql SELECT product_code, warehouse_code, quantity, unit, update_time FROM stock_balance df pd.read_sql(sql, conn) conn.close() return df def clean_data(df): # 1. 编码归一化 df[product_code] df[product_code].apply(normalize_code) df[warehouse_code] df[warehouse_code].apply(normalize_code) # 2. 统一单位并计算标准数量 clean_rows [] for _, row in df.iterrows(): qty, remark convert_quantity(row[quantity], row[unit]) clean_rows.append({ product_code: row[product_code], warehouse_code: row[warehouse_code], quantity: qty, unit: 件, update_time: row[update_time], }) clean_df pd.DataFrame(clean_rows) # 3. 按商品 仓库去重保留最新时间 clean_df clean_df.sort_values(update_time, ascendingFalse) clean_df clean_df.drop_duplicates(subset[product_code, warehouse_code], keepfirst) # 4. 过滤空编码 clean_df clean_df[(clean_df[product_code] ! ) (clean_df[warehouse_code] ! )] return clean_df def write_target_data(df): conn pymysql.connect(**TARGET_DB_CONFIG) cursor conn.cursor() insert_sql INSERT INTO stock_balance_new (product_code, warehouse_code, quantity, unit, update_time) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE quantity VALUES(quantity), unit VALUES(unit), update_time VALUES(update_time) total 0 batch_size 500 batch [] for _, row in df.iterrows(): batch.append(( row[product_code], row[warehouse_code], row[quantity], row[unit], row[update_time], )) if len(batch) batch_size: cursor.executemany(insert_sql, batch) conn.commit() total len(batch) print(f[{datetime.now()}] 已写入 {total} 条) batch.clear() if batch: cursor.executemany(insert_sql, batch) conn.commit() total len(batch) print(f[{datetime.now()}] 已写入 {total} 条) cursor.close() conn.close() print(迁移完成总计写入:, total) def main(): print(步骤1读取老库数据) df fetch_source_data() print(读取到数据行数:, len(df)) print(步骤2数据清洗) clean_df clean_data(df) print(清洗后数据行数:, len(clean_df)) print(步骤3写入新库) write_target_data(clean_df) if __name__ __main__: main()5.3 代码说明fetch_source_data从老库一次性读取全部库存余额。如果数据量特别大建议增加LIMIT分批次读取不要一次全部加载到内存。clean_data完成编码归一化、单位换算、去重、空值过滤。write_target_data分批写入新库并打印进度日志。使用ON DUPLICATE KEY UPDATE实现重复写入时更新而不是报错。增加batch_size分批提交避免大批量插入时占用过多事务资源。这里要特别提醒一句这个脚本是演示思路实际迁移前一定要先在测试库完整跑一遍确认清洗规则和写入逻辑没问题再对生产库执行。5.4 运行结果正常执行后命令行输出大致如下步骤1读取老库数据 读取到数据行数: 18632 步骤2数据清洗 清洗后数据行数: 17354 步骤3写入新库 [2025-01-15 10:23:45] 已写入 500 条 [2025-01-15 10:23:45] 已写入 1000 条 ... 迁移完成总计写入: 17354清洗前 18632 条清洗后 17354 条说明有 1278 条是重复或者无效数据。这个数字本身就是迁移质量的一个指标。6. 数据一致性核对与验证迁移完成不等于结束。真正的考验是对账。6.1 数量级核对最保守的做法是在迁移后分别对老库和新库执行几个统计 SQL对比结果。-- 老库总商品数、总库存量、有负库存的商品数 SELECT COUNT(*) AS total_rows, SUM(quantity) AS total_qty, SUM(CASE WHEN quantity 0 THEN 1 ELSE 0 END) AS negative_cnt FROM old_db.stock_balance;-- 新库同样的统计 SELECT COUNT(*) AS total_rows, SUM(quantity) AS total_qty, SUM(CASE WHEN quantity 0 THEN 1 ELSE 0 END) AS negative_cnt FROM new_db.stock_balance_new;注意单位换算后两边总数量可能不同不能直接拿总数做等值比较。正确的做法是先把老库数量按同一单位换算后再对比。也就是说“老库总件数”应该等于“新库总件数”因为单位换算只是乘以固定倍数但总量本身不应该发生业务意义上的变化。6.2 逐商品核对可以用一条 SQL 把两边的商品维度数据关联起来SELECT o.product_code, o.old_qty, n.new_qty, (n.new_qty - o.old_qty) AS diff_qty FROM ( SELECT product_code, SUM(quantity) AS old_qty FROM old_db.stock_balance GROUP BY product_code ) o LEFT JOIN ( SELECT product_code, SUM(quantity) AS new_qty FROM new_db.stock_balance_new GROUP BY product_code ) n ON o.product_code n.product_code WHERE (n.new_qty - o.old_qty) 0 OR n.product_code IS NULL LIMIT 100;这条 SQL 能直接找出“老库有但新库没有”的产品以及“两边数量不一致”的产品。检查结果后再决定是否需要手工修正。6.3 抽样核对数据量很大时全量核对耗时太长可以改为分层抽样。按商品编码分布随机抽取 5% 的商品逐一核对老库流水、新库余额、仓库实物盘点数量三个数字是否对得上。对账结果建议输出成一张核对报告核对项结果老库记录总数18632新库记录总数17354单位换算后总件数差异0存在差异的商品数0无法匹配商品数0负库存迁移数12业务已确认如果报告中任何一项不是预期结果都要先暂停业务切换回到清洗和迁移脚本里排查。7. 常见问题与排查思路库存迁移项目里的坑不少是重复出现的。这里列一个高频问题表方便实操时对应排查。问题现象常见原因解决思路迁移后部分商品在新库查不到老库编码包含不可见字符清洗时没有被过滤掉查看编码的 HEX 值去除前后空格、换行、零宽字符库存总数对不上单位换算错误某类商品用了错误倍率单独列出该商品规格与业务确认最小库存单位后重新换算同一个商品多条记录老库本身存在重复数据或新表没有唯一键加唯一键迁移前先按商品仓库去重负数库存导致后续出库计算异常老库负库存未处理直接迁移迁移前由业务确认保留标记或按规则调整插入速度很慢没有分批提交或者一次插入的数据量过大使用executemany 批量 commit控制批次大小迁移脚本中途报错数据重复写入脚本不支持断点续跑异常后从头执行增加批次状态表记录已迁移的商品编码范围新库执行 SQL 时死锁多个迁移任务同时写入同一张表串行执行迁移任务或者使用INSERT ... ON DUPLICATE KEY UPDATE迁移后发现仓库编码失效老仓库已停用但库存数据仍然存在和业务确认停用仓库是否迁移不迁移则单独导出归档排查时建议按这个顺序走先确认数据量是否一致再确认编码映射是否完整然后检查清洗规则有没有漏掉特殊数据最后检查写入脚本是否被重复执行。8. 最佳实践与工程建议迁移项目做完一遍之后可以沉淀出下面几条通用经验放在下一个项目里直接复用。第一迁移脚本必须可重跑、可回滚。写脚本的时候把“重复执行”当成默认场景来处理。写入目标表之前先备份目标表或者把数据放到临时表确认无误后再合并进正式表。这样即使清洗规则写错了也能快速回滚重来。第二迁移过程要输出日志。每个批次写入多少条、清洗掉了多少条、哪些数据被标记为异常都要落盘记录下来。迁移不是“跑完就结束”而是“跑完还能复盘”。没有日志的迁移出了问题只能靠猜。第三先小范围试点再全量执行。建议先抽取一个仓库或一个商品分类的数据做完整迁移和核对确认流程没问题后再放开全量。试点阶段发现的问题通常能覆盖全量阶段 80% 的坑。第四备份不能省。无论迁移方案设计得多完美都要在迁移前一天对老库做完整备份对目标新库也做一次快照。备份不是形式是你最后的退路。第五迁移期间禁止业务侧同时修改库存。如果没法完全停机至少要做增量同步。不要让业务数据一边写老库、一边让迁移脚本全量覆盖这样必然产生脏数据。第六权限最小化。迁移脚本使用的数据库账号只给需要操作的库表权限不要直接给 root 或管理员权限。生产环境变更前脚本要经过代码评审执行时要有人确认、有操作记录做到可追溯。9. 总结从“古早库存”和“搬家之前的库存”这两句话延伸出来我们完整梳理了库存数据迁移的整个流程先盘点和清洗数据再设计迁移方案然后编写迁移脚本最后用对账验证结果。在实操中真正花时间的往往不是写代码那一两个小时而是数据清洗和后续核对。编码不统一、单位不统一、重复数据、负库存这些问题在老系统里几乎必然存在提前做好清洗规则迁移才能顺利进行。如果你接下来也要做类似的数据迁移项目建议先画一张简单的迁移流程图把数据流、责任人、校验节点标清楚再动手写代码。库存数据直接关系到仓管、财务、采购和销售任何一条数据错了都可能引发线下问题所以迁移前的备份、迁移中的日志、迁移后的对账这三件事一件都不能少。把这套思路消化掉下次再遇到“搬家前的库存”你就知道该怎么处理了。如果这篇文章对你有帮助可以收藏备用。等真正做数据迁移项目的时候再翻出来对照排查。