PostgreSQL含生成列的多表批量插入方案优化问询

针对你每秒新增约5000条、初始还有5万条转储数据的场景，当前Scala流程里那种先插location再全量查询、替换ID，再处理phone的方式，在高并发下肯定会遇到性能瓶颈——毕竟来回数据库的次数太多了。下面我结合PostgreSQL的特性，给你一套能大幅减少数据库交互、高效处理多表批量插入+关联映射的优化方案：

一、用INSERT ... ON CONFLICT合并插入+ID查询操作 #

PostgreSQL的UPSERT（ON CONFLICT）特性可以在插入时自动处理重复数据，同时配合RETURNING子句，一次性获取所有插入/匹配记录的生成ID，直接替代你现在“插入+全量查询”的两次冗余操作。

1. Location表批量处理示例 #

假设你的location表有唯一约束（避免重复地址）：

CREATE TABLE locations (
    id SERIAL PRIMARY KEY,
    address TEXT UNIQUE NOT NULL
);

你可以在Scala里一次性批量插入所有地址，同时直接拿到每个地址对应的ID映射：

// 先对地址去重，减少数据库处理量
val uniqueAddresses = persons.map(_.location).distinct

// 批量插入+返回ID的SQL
val insertLocationSql = """
    INSERT INTO locations (address)
    VALUES (?)
    ON CONFLICT (address) DO NOTHING  -- 重复地址不插入，直接返回已有ID
    RETURNING id, address
"""

// 通过JDBC/Slick等框架执行批量操作，获取(id, address)映射
val locationIdMap = executeBatchReturning(insertLocationSql, uniqueAddresses).toMap

这一步就完成了“去重插入+ID映射获取”，完全不需要额外的全量查询。

2. Phone Number表同理处理 #

对phone number表套用同样逻辑，先确保表有唯一约束：

CREATE TABLE phone_numbers (
    id SERIAL PRIMARY KEY,
    number TEXT UNIQUE NOT NULL
);

批量插入并获取ID映射：

val uniquePhones = persons.flatMap(_.phones).distinct
val insertPhoneSql = """
    INSERT INTO phone_numbers (number)
    VALUES (?)
    ON CONFLICT (number) DO NOTHING
    RETURNING id, number
"""
val phoneIdMap = executeBatchReturning(insertPhoneSql, uniquePhones).toMap

二、批量插入主表+关联表 #

拿到location和phone的ID映射后，就可以批量插入主表（Persons）和关联表（如果是多对多关系）。

1. 主表Persons批量插入 #

假设Persons表结构如下：

CREATE TABLE persons (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    location_id INT NOT NULL REFERENCES locations(id),
    -- 其他业务字段
);

批量插入代码：

// 把原始person数据中的location替换为对应的ID
val personData = persons.map(p => (p.name, locationIdMap(p.location)))

val insertPersonSql = """
    INSERT INTO persons (name, location_id)
    VALUES (?, ?)
    RETURNING id, name
"""
// 可选：如果需要后续关联phone，保存person的ID映射
val personIdMap = executeBatchReturning(insertPersonSql, personData).toMap

2. 多对多关联表批量插入（Phone-Person） #

如果存在person_phones关联表：

CREATE TABLE person_phones (
    person_id INT REFERENCES persons(id),
    phone_id INT REFERENCES phone_numbers(id),
    PRIMARY KEY (person_id, phone_id)  -- 避免重复关联
);

批量生成关联数据并插入：

val personPhonePairs = persons.flatMap(p => 
    p.phones.map(phone => (personIdMap(p.name), phoneIdMap(phone)))
)

val insertRelSql = """
    INSERT INTO person_phones (person_id, phone_id)
    VALUES (?, ?)
    ON CONFLICT DO NOTHING  -- 跳过已存在的关联
"""
executeBatch(insertRelSql, personPhonePairs)

三、超大批量数据：用COPY命令提速 #

对于初始的5万条转储数据，或者每秒5000条的峰值流量，PostgreSQL的COPY命令比普通批量插入性能提升数倍——它是专门为大批量数据导入设计的高效方式。

操作流程示例（以location表为例） #

-- 1. 创建临时表，结构和正式表匹配（只保留需要导入的字段）
CREATE TEMP TABLE temp_locations (address TEXT NOT NULL);

-- 2. 用CopyManager导入数据（Scala里可以通过PostgreSQL的JDBC扩展实现）
COPY temp_locations FROM STDIN WITH (FORMAT csv);

-- 3. 同步到正式表并获取ID
INSERT INTO locations (address)
SELECT address FROM temp_locations
ON CONFLICT (address) DO NOTHING
RETURNING id, address;

Scala中可以用org.postgresql.copy.CopyManager来高效写入CSV格式的批量数据，避免单条插入的网络开销。

关键优化点总结 #

• 减少数据库往返：用INSERT ... ON CONFLICT ... RETURNING把“插入+查询”合并为一次操作，砍掉不必要的IO开销。
• 应用层去重：先对location、phone数据去重，减少数据库端的重复处理压力。
• 全流程批量：所有插入操作都用批量执行，避免单条SQL的网络往返。
• 超大批量用COPY：初始转储或峰值流量时，用COPY替代普通批量插入，最大化导入效率。

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