跨两台服务器的四张数据库表迁移合并脚本开发需求

刚入职就碰到跨库跨服务器的表合并需求？我之前处理过几乎一模一样的场景，给你整理一套实用的脚本开发思路和示例，帮你快速落地：

一、列完全匹配的Table A → Table A2 迁移 #

这部分是最省心的，因为结构完全一致，核心就是处理重复数据（毕竟是合并，不是覆盖）：

• 核心思路：先获取目标表已有的主键数据，筛选源表中目标表没有的记录，再批量插入。如果需要更新已有数据，优先用INSERT ... ON DUPLICATE KEY UPDATE（MySQL）或ON CONFLICT DO UPDATE（PostgreSQL），避免直接删除数据导致丢失。
• Python脚本示例（通用型强，支持多种数据库）：
  我常用Python+SQLAlchemy+pandas来做这种操作，代码可读性高，适配MySQL、PostgreSQL、SQL Server都没问题：

  from sqlalchemy import create_engine
  import pandas as pd
  import logging
  
  # 配置日志，方便排查问题
  logging.basicConfig(level=logging.INFO)
  logger = logging.getLogger(__name__)
  
  # 源/目标数据库连接字符串（根据你的数据库类型调整）
  source_db_url = 'mysql+pymysql://用户名:密码@源服务器IP:端口/源数据库名'
  target_db_url = 'mysql+pymysql://用户名:密码@目标服务器IP:端口/目标数据库名'
  
  # 创建数据库连接引擎
  source_engine = create_engine(source_db_url)
  target_engine = create_engine(target_db_url)
  
  try:
      # 读取源表数据
      source_a_data = pd.read_sql_table('TableA', source_engine)
      logger.info(f"成功读取源表TableA的{len(source_a_data)}条数据")
  
      # 获取目标表已有的主键（假设主键是id，根据实际调整）
      target_a_ids = pd.read_sql("SELECT id FROM TableA2", target_engine)['id'].tolist()
  
      # 筛选目标表没有的新数据
      new_data = source_a_data[~source_a_data['id'].isin(target_a_ids)]
  
      if not new_data.empty:
          # 批量插入目标表
          new_data.to_sql('TableA2', target_engine, if_exists='append', index=False)
          logger.info(f"成功向TableA2插入{len(new_data)}条新数据")
      else:
          logger.info("TableA没有新数据需要迁移")
  except Exception as e:
      logger.error(f"处理TableA迁移时出错：{str(e)}")
  

• 备选方案（数据库原生工具）：如果不想写代码，用数据库自带工具更快。比如MySQL可以用mysqldump导出源表，再用mysql命令导入，加上--insert-ignore跳过重复数据：

  # 导出源TableA
  mysqldump -h 源服务器IP -u 用户名 -p 源数据库名 TableA > table_a.sql
  # 导入到目标TableA2（跳过重复主键）
  mysql -h 目标服务器IP -u 用户名 -p 目标数据库名 --local-infile=1 -e "INSERT IGNORE INTO TableA2 SELECT * FROM TableA;"
  

二、列部分相似的Table B → Table B2 迁移 #

这部分需要先理清楚列的映射关系，是关键步骤：

1. 先梳理列映射表：把两张表的列一一对应，标记出需要转换、设置默认值或者丢弃的列，比如：

   源Table B列	目标Table B2列	处理逻辑
   user_id	id	主键，用于去重
   real_name	full_name	直接映射
   register_date	join_time	时间格式转换（比如字符串转datetime）
   -	account_status	设置默认值1（正常状态）
   unused_col	-	丢弃该列

2. Python脚本示例：基于上面的映射表，做列转换和数据插入：

try:
# 读取源表TableB数据
source_b_data = pd.read_sql_table('TableB', source_engine)
logger.info(f"成功读取源表TableB的{len(source_b_data)}条数据")

# 获取目标表已有的主键
  target_b_ids = pd.read_sql("SELECT id FROM TableB2", target_engine)['id'].tolist()
  # 筛选新数据
  new_data_b = source_b_data[~source_b_data['user_id'].isin(target_b_ids)]

  if not new_data_b.empty:
      # 1. 列重命名（映射）
      new_data_b = new_data_b.rename(columns={
          'user_id': 'id',
          'real_name': 'full_name',
          'register_date': 'join_time'
      })
      # 2. 添加目标表需要的新列，设置默认值
      new_data_b['account_status'] = 1
      # 3. 时间格式转换（如果需要）
      new_data_b['join_time'] = pd.to_datetime(new_data_b['join_time'], format='%Y-%m-%d')
      # 4. 丢弃不需要的列
      new_data_b = new_data_b.drop(columns=['unused_col'])

      # 插入目标表
      new_data_b.to_sql('TableB2', target_engine, if_exists='append', index=False)
      logger.info(f"成功向TableB2插入{len(new_data_b)}条新数据")
  else:
      logger.info("TableB没有新数据需要迁移")

except Exception as e:
logger.error(f"处理TableB迁移时出错：{str(e)}")

3. **关键注意事项**：
 - 一定要先拿**小批量测试数据**跑一遍，验证列转换、默认值设置是否正确，避免全量插入出错
 - 如果有复杂的业务逻辑（比如源端的枚举值要转成目标端的另一个），可以用`map`函数处理：
   ```python
   status_map = {'active': 1, 'inactive': 0}
   new_data_b['account_status'] = new_data_b['source_status'].map(status_map)
   ```

## 三、脚本优化建议
- **增量迁移**：如果后续需要定期同步，不要每次全量读取，而是基于时间戳（比如`WHERE update_time > 上次同步时间`）或者自增主键做增量查询，大幅提高效率
- **事务处理**：如果是多表迁移，开启数据库事务，确保要么全部成功，要么回滚，避免数据不一致
- **日志完善**：把日志输出到文件，方便后续排查问题，比如添加`fileHandler`到logging配置

内容的提问来源于stack exchange，提问作者Black-Prince