从MySQL迁移128GB数据到PostgreSQL后约束异常咨询

提供的相关信息 #

约束查询输出 #

(33129, 'forecasts_pkey', 2200, 'p', False, False, True, 33124, 0, 33128, 0, 0, ' ', ' ', ' ', True, 0, True, [1], None, None, None, None, None, None)
(33176, 'fk_models', 2200, 'f', False, False, True, 33124, 0, 33115, 0, 33112, 'a', 'a', 's', True, 0, True, [2], [1], [96], [96], [96], None, None)
(33182, 'unique_seriesid_modelrundate_modelrun_valuetimeutc_publishedatu', 2200, 'u', False, False, True, 33124, 0, 33181, 0, 0, ' ', ' ', ' ', True, 0, True, [2, 3, 7, 5, 8], None, None, None, None, None, None)

建表脚本 #

create table if not exists ModelType (
    Id int primary key,
    ModelType varchar (50)
);

create table if not exists Models (
    SeriesId int primary key,
    model varchar(50),
    ModelGroup varchar(50),
    Type varchar(20),
    Country varchar(50),
    Provider varchar(50),
    WeatherSystem varchar(10),
    Unit varchar(50),
    Area Varchar(10),
    ModelTypeId int,
    constraint fk_modeltype
        Foreign key (ModelTypeId) 
        references ModelType (Id)
);

CREATE TABLE IF NOT EXISTS Forecasts (
    id Serial PRIMARY KEY,
    SeriesId int,
    ModelRunDate date,
    InsertedAtUTC timestamp,
    ValueTimeUTC timestamp,
    Value REAL,
    ModelRun int,
    PublishedAtUTC timestamp,
);

数据加载后执行的脚本 #

def delete_duplicate_rows_from_forecast():
    cur = conn.cursor()
    sql = '''DELETE FROM forecasts f1
                    USING forecasts f2
                    WHERE f1.ctid < f2.ctid
                    AND f1.seriesid = f2.seriesid
                    AND f1.modelrundate = f2.modelrundate
                    AND f1.modelrun = f2.modelrun
                    AND f1.valuetimeutc = f2.valuetimeutc
                    AND f1.insertedatutc = f2.insertedatutc;'''
    cur.execute(sql)
    cur.close()
    conn.commit()

def set_constraints_after_load():
    delete_duplicate_rows_from_forecast()
    cur = conn.cursor()
    try:
        cur.execute('''ALTER TABLE Forecasts
            ADD CONSTRAINT fk_models
                FOREIGN KEY (SeriesId) 
                REFERENCES Models (SeriesId);''')
        cur.execute('''ALTER TABLE Forecasts
            ADD CONSTRAINT unique_seriesid_modelrundate_modelrun_valuetimeutc_publishedatutc
            UNIQUE (seriesid,modelrundate, modelrun,valuetimeutc, publishedatutc);''')
        conn.commit()

    except Exception as e:
        print(f'Set constraints error: {e}')
    finally:
        cur.close
        conn.close()

获取所有约束的脚本 #

def get_constraints(tablename):
    cur = conn.cursor()
    sql = f'''SELECT con.*
        FROM pg_catalog.pg_constraint con
        INNER JOIN pg_catalog.pg_class rel ON rel.oid = con.conrelid
        INNER JOIN pg_catalog.pg_namespace nsp ON nsp.oid = connamespace
        WHERE  rel.relname = '{tablename}';'''
    cur.execute(sql)
    for record in cur:
        print(record)

关于“奇怪约束”的问题 #

你看到的数字、布尔值组成的约束信息，是直接查询pg_catalog.pg_constraint系统表返回的原始元数据，这个表存储的是PostgreSQL约束的底层属性，并非友好的展示形式。

其中几个关键字段的含义：

• 第二个值是约束名称（比如forecasts_pkey、fk_models）
• 第四个值是约束类型：p代表主键，f代表外键，u代表唯一约束，这和你预期的约束类型完全一致。

如果要查看清晰的约束信息，改用information_schema.table_constraints视图查询即可，修改后的脚本如下：

def get_constraints(tablename):
    cur = conn.cursor()
    sql = f'''
        SELECT constraint_name, constraint_type
        FROM information_schema.table_constraints
        WHERE table_name = '{tablename}'
        AND constraint_schema = 'public';
    '''
    cur.execute(sql)
    for record in cur:
        print(record)

关于128GB数据后添加唯一约束的可行性 #

完全可行，但需要注意以下几点：

1. 优化重复数据删除效率：你现有删除重复数据的脚本对大表来说扫描次数过多，可改用窗口函数批量删除，减少IO开销：

   WITH duplicates AS (
       SELECT ctid,
              ROW_NUMBER() OVER (PARTITION BY seriesid, modelrundate, modelrun, valuetimeutc, publishedatutc ORDER BY ctid) AS rn
       FROM forecasts
   )
   DELETE FROM forecasts
   WHERE ctid IN (SELECT ctid FROM duplicates WHERE rn > 1);
   

2. 避免长时间锁表：直接添加约束会触发全表扫描并持有排他锁，阻塞业务读写。可分两步操作：

   -- 第一步：快速添加约束，不验证现有数据，锁表时间极短
   ALTER TABLE Forecasts
   ADD CONSTRAINT unique_seriesid_modelrundate_modelrun_valuetimeutc_publishedatutc
   UNIQUE (seriesid,modelrundate, modelrun,valuetimeutc, publishedatutc) NOT VALID;
   
   -- 第二步：在业务低峰期验证现有数据，此过程不阻塞写入
   VALIDATE CONSTRAINT unique_seriesid_modelrundate_modelrun_valuetimeutc_publishedatutc;
   

3. 提升执行性能：临时调大work_mem参数，让排序、去重操作在内存中完成，避免磁盘临时文件开销：

   SET work_mem = '2GB'; -- 根据服务器内存调整，比如16G内存服务器可设为2-4GB
   

内容的提问来源于stack exchange，提问作者Chriscolle