企业官网建设流程全解析

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 Server，SQL 语法层面的兼容需要另外调整，但迁移的思路是通用的。

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", ascending=False) df = df.drop_duplicates(subset=["product_code", "warehouse_code"], keep="first")

4. 库存数据迁移方案设计

清洗完成之后，进入迁移方案设计。这个环节决定上线时怎么切，也是整个迁移项目中最核心的一步。

4.1 方案A：全量停机迁移

适合业务量不大、允许短时间停机的场景。流程如下：

1. 业务停止操作。
2. 备份老库数据。
3. 执行全量迁移脚本，把清洗后的数据写入新库。
4. 执行对账，确认数量一致。
5. 切换业务系统到新库。

优点：逻辑简单，容易核对。 缺点：停机时间长，业务影响范围大。

4.2 方案B：全量迁移 + 增量同步

适合数据量大、停机时间不能太长的场景。流程如下：

1. 先跑一次全量迁移。
2. 在迁移过程中，记录老库业务操作的日志表或时间戳。
3. 全量迁移完成后，再跑一次增量同步，把迁移期间产生的变化同步到新库。
4. 最后短暂停机，做最终一致性校验后切换。

这种方式比停机迁移复杂，但对大库存、连续生产业务的场景友好很多。

4.3 方案C：双写

双写是在新系统正式切换前，让业务同时写入老库和新库，运行一段时间确认新库稳定后再下线老库。

双写的问题在于：如果两边代码事务不统一，很容易造成数据不一致，调试成本较高。很多小团队并不适合直接上双写，一般建议先用“全量 + 增量”或者“停机迁移”解决。

4.4 迁移工具选型

如果数据量不大，直接用 Python 脚本 + SQL 就够了。如果数据量达到百万级以上，建议用成熟的 ETL 工具（如 DataX、Kettle），或者在数据库层面用INSERT ... SELECT的方式做批量迁移。

选择工具的核心原则是：可回滚、可重跑、有日志。不管用什么工具，迁移脚本必须支持断点续跑，不能跑到一半失败就全部从头再来。

5. 实战案例：Python 实现库存迁移

下面用一个最小可运行的示例，演示从老库读取库存数据、清洗、写入新库的完整过程。为了便于理解，我们使用两张表：

老库old_db.stock_balance：

CREATE 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`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

新库new_db.stock_balance_new：

CREATE 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`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

新表增加了一个唯一键，避免迁移后同一个商品 + 仓库出现多条记录。

5.1 安装依赖

pip install pymysql pandas

5.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", ascending=False) clean_df = clean_df.drop_duplicates(subset=["product_code", "warehouse_code"], keep="first") # 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% 的商品，逐一核对老库流水、新库余额、仓库实物盘点数量，三个数字是否对得上。

对账结果建议输出成一张核对报告：

如果报告中任何一项不是预期结果，都要先暂停业务切换，回到清洗和迁移脚本里排查。

7. 常见问题与排查思路

库存迁移项目里的坑，不少是重复出现的。这里列一个高频问题表，方便实操时对应排查。

排查时建议按这个顺序走：先确认数据量是否一致，再确认编码映射是否完整，然后检查清洗规则有没有漏掉特殊数据，最后检查写入脚本是否被重复执行。

8. 最佳实践与工程建议

迁移项目做完一遍之后，可以沉淀出下面几条通用经验，放在下一个项目里直接复用。

第一，迁移脚本必须可重跑、可回滚。写脚本的时候，把“重复执行”当成默认场景来处理。写入目标表之前，先备份目标表或者把数据放到临时表，确认无误后再合并进正式表。这样即使清洗规则写错了，也能快速回滚重来。

第二，迁移过程要输出日志。每个批次写入多少条、清洗掉了多少条、哪些数据被标记为异常，都要落盘记录下来。迁移不是“跑完就结束”，而是“跑完还能复盘”。没有日志的迁移，出了问题只能靠猜。

第三，先小范围试点，再全量执行。建议先抽取一个仓库或一个商品分类的数据做完整迁移和核对，确认流程没问题后，再放开全量。试点阶段发现的问题，通常能覆盖全量阶段 80% 的坑。

第四，备份不能省。无论迁移方案设计得多完美，都要在迁移前一天对老库做完整备份，对目标新库也做一次快照。备份不是形式，是你最后的退路。

第五，迁移期间禁止业务侧同时修改库存。如果没法完全停机，至少要做增量同步。不要让业务数据一边写老库、一边让迁移脚本全量覆盖，这样必然产生脏数据。

第六，权限最小化。迁移脚本使用的数据库账号，只给需要操作的库表权限，不要直接给 root 或管理员权限。生产环境变更前，脚本要经过代码评审，执行时要有人确认、有操作记录，做到可追溯。

9. 总结

从“古早库存”和“搬家之前的库存”这两句话延伸出来，我们完整梳理了库存数据迁移的整个流程：先盘点和清洗数据，再设计迁移方案，然后编写迁移脚本，最后用对账验证结果。在实操中，真正花时间的往往不是写代码那一两个小时，而是数据清洗和后续核对。编码不统一、单位不统一、重复数据、负库存，这些问题在老系统里几乎必然存在，提前做好清洗规则，迁移才能顺利进行。

如果你接下来也要做类似的数据迁移项目，建议先画一张简单的迁移流程图，把数据流、责任人、校验节点标清楚，再动手写代码。库存数据直接关系到仓管、财务、采购和销售，任何一条数据错了都可能引发线下问题，所以迁移前的备份、迁移中的日志、迁移后的对账，这三件事一件都不能少。

把这套思路消化掉，下次再遇到“搬家前的库存”，你就知道该怎么处理了。如果这篇文章对你有帮助，可以收藏备用。等真正做数据迁移项目的时候，再翻出来对照排查。