PostgreSQL 数据迁移时数据丢失问题排查与解决-Python教程-PHP中文网

PostgreSQL 数据迁移时数据丢失问题排查与解决

DDD

发布： 2025-08-23 20:08:01

原创

735人浏览过

postgresql 数据迁移时数据丢失问题排查与解决

本文针对PostgreSQL数据库之间数据迁移过程中出现的数据丢失问题，提供详细的排查思路和解决方案。通过分析连接配置、SQL脚本执行逻辑以及潜在的数据库连接混淆问题，帮助开发者避免类似错误，确保数据迁移的准确性和完整性。重点关注.env配置文件，以及DROP TABLE语句可能带来的风险，并提供相应的代码示例和最佳实践。

在进行数据库迁移时，数据丢失是一个严重的问题。以下将深入分析可能导致数据丢失的原因，并提供相应的解决方案。

确认数据库连接配置

最常见的原因是连接配置错误。在你的connect.py文件中，使用了.env文件来存储数据库连接信息。务必仔细检查zno.env和zno_norm.env这两个文件，确保它们指向不同的数据库。尤其要确认以下几点：

DB_DATABASE: 确保两个.env文件中的DB_DATABASE变量分别指向zno_db和zno_db_norm。
DB_HOST, DB_PORT, DB_USERNAME, DB_PASSWORD: 检查这些参数是否正确，特别是当两个数据库位于同一服务器上时。

如果两个.env文件中的DB_DATABASE指向同一个数据库，那么你的代码实际上是在同一个数据库上执行了两次相同的操作，这会导致数据被覆盖或删除。

代码示例

假设zno.env文件内容如下：

DB_USERNAME=user1
DB_PASSWORD=password1
DB_DATABASE=zno_db
DB_HOST=localhost
DB_PORT=5432

登录后复制

zno_norm.env文件内容应该如下：

DB_USERNAME=user1
DB_PASSWORD=password1
DB_DATABASE=zno_db_norm
DB_HOST=localhost
DB_PORT=5432

登录后复制

请务必核对这两个文件的内容，确保DB_DATABASE变量的值不同。

分析 SQL 脚本中的 DROP TABLE 语句

2_NORMAL_TABLE.sql脚本中包含DROP TABLE IF EXISTS语句，这可能是导致数据丢失的另一个原因。

DROP TABLE IF EXISTS educate_organisation, participants, loc, test, participants_eo;

登录后复制

这段代码会删除指定的表，如果脚本被错误地执行在错误的数据库上，会导致数据丢失。

解决方案：

怪兽AI数字人

数字人短视频创作，数字人直播，实时驱动数字人

查看详情

谨慎使用 DROP TABLE: 在生产环境中，应尽量避免使用DROP TABLE语句。如果必须使用，请务必仔细检查执行环境，确保操作的是正确的数据库。
备份数据: 在执行任何可能导致数据丢失的操作之前，务必备份数据。
事务控制: 将创建表的操作放在一个事务中，如果创建失败，可以回滚事务，避免数据不一致。

修改后的 SQL 脚本示例

为了更安全地创建表，可以考虑以下做法：

-- 不删除表，而是先检查表是否存在
CREATE TABLE IF NOT EXISTS loc (
    locid SERIAL PRIMARY KEY,
    locname varchar,
    locregname varchar,
    locareaname varchar,
    loctername varchar,
    UNIQUE(locname, locregname, loctername)
);

CREATE TABLE IF NOT EXISTS educate_organisation (
    eoid SERIAL PRIMARY KEY,
    eoname varchar,
    eoregname varchar,
    eoareaname varchar,
    eotername varchar,
    eoparent varchar,
    eotypename varchar,
    locid INTEGER,
    FOREIGN KEY (locid) REFERENCES loc (locid),
    UNIQUE(eoname, eotypename, eoparent)
);

-- 其他表的创建语句类似

登录后复制

检查代码执行流程

你的main.py文件控制了整个数据迁移的流程。请仔细检查代码的执行顺序，确保每个步骤都在正确的数据库上执行。

from table import table
from insert import insert_data, migrate_data
from result import get_result

if __name__ == "__main__":
     table("sql/1_CREATE_TABLE.sql", "config/zno.env", "Create table zno\n")
     insert_data()
     get_result()

     table("sql/2_NORMAL_TABLE.sql", "config/zno.env", "Norming tables\n")
     table("sql/3_MIGRATION.sql", "config/zno.env", "Migrations\n")
     table("sql/2_NORMAL_TABLE.sql", "config/zno_norm.env", "Creating second bd\n")

登录后复制

从代码中可以看出，2_NORMAL_TABLE.sql首先在zno.env指定的数据库上执行，然后再在zno_norm.env指定的数据库上执行。如果zno.env和zno_norm.env指向同一个数据库，那么DROP TABLE语句会先删除zno_db中的表，然后再创建新的表，导致数据丢失。

解决方案：

使用不同的连接对象: 确保在操作不同的数据库时，使用不同的连接对象。
显式指定数据库: 在 SQL 语句中显式指定数据库名称，避免混淆。

优化连接管理

你的connect.py文件中的get_conn函数使用了简单的重试机制。虽然这可以处理一些连接问题，但可能会掩盖更深层次的错误。

import psycopg
import time
from dotenv import load_dotenv
import os

def get_conn(envpath="config/zno.env"):
     load_dotenv(envpath)

     username = os.getenv("DB_USERNAME")
     password = os.getenv("DB_PASSWORD")
     database = os.getenv("DB_DATABASE")
     host = os.getenv("DB_HOST")
     port = os.getenv("DB_PORT")

     try:
          conn = psycopg.connect(user=username, password=password, dbname=database, host=host, port=port)
          return conn
     except:
          print("Reconnecting...")
          time.sleep(5)
          get_conn()

登录后复制

建议改进：

更详细的错误处理: 在except块中，记录更详细的错误信息，例如异常类型和错误消息。
限制重试次数: 避免无限重试，设置最大重试次数，并在达到最大次数后抛出异常。
使用连接池: 对于高并发的应用程序，使用连接池可以提高性能和可靠性。

改进后的代码示例

import psycopg
import time
import os
from dotenv import load_dotenv

def get_conn(envpath="config/zno.env", max_retries=3):
    load_dotenv(envpath)

    username = os.getenv("DB_USERNAME")
    password = os.getenv("DB_PASSWORD")
    database = os.getenv("DB_DATABASE")
    host = os.getenv("DB_HOST")
    port = os.getenv("DB_PORT")

    retries = 0
    while retries < max_retries:
        try:
            conn = psycopg.connect(user=username, password=password, dbname=database, host=host, port=port)
            return conn
        except psycopg.Error as e:
            print(f"Connection failed: {e}")
            retries += 1
            time.sleep(5)
    raise Exception("Failed to connect to the database after multiple retries.")

登录后复制