如何优化PostgreSQL数据库的批量插入性能？

问题背景 #

现有C#服务通过Dapper调用Postgres存储过程处理客户数据，逻辑如下：

• 若客户已存在，获取其GUID并插入/更新CustomerInformations表
• 若客户不存在，先插入Customers表获取GUID，再处理CustomerInformations表

当前CustomerInformations表已有7500万条记录，插入速度从之前的175万条/小时降至20万条/小时。无法修改现有数据存储方式，当前逻辑为循环为每个客户属性调用存储过程，单次调用可能包含两次插入操作。

现有代码 #

C#服务代码 #

foreach (var info in request.Data)
{
    string sql = "add_one_by_customer";
    object parameters = new
    {
        p_customer_first_name = info.FirstName,
        p_customer_last_name = info.LastName,
        p_customer_property_name = info.PropertyName,
        p_customer_property_value = info.PropertyValue
    };

    try
    {
        await db.ExecuteAsync(sql, parameters, transaction: transaction, commandType: CommandType.StoredProcedure);
    }
    catch (Exception e)
    {
        throw new Exception($"Failed to insert");
    }
}

Postgres存储过程 #

CREATE OR REPLACE PROCEDURE add_one_by_customer(
    p_customer_first_name  VARCHAR,
    p_customer_last_name  VARCHAR,
    p_customer_property_name  VARCHAR,
    p_customer_property_value  VARCHAR
)
LANGUAGE plpgsql
AS $procedure$
DECLARE 
    p_customer_id uuid;
    p_current_item_value varchar;   
begin   
    -- 注：原语句存在列数不匹配错误，应为SELECT customer_id INTO p_customer_id
    SELECT INTO p_customer_id,
                customer_id
    FROM customers
    WHERE customer_first_name = p_customer_first_name AND
          customer_last_name = p_customer_last_name
    limit 1;
                           
    
    IF (p_customer_id IS NULL) THEN  
        begin               
            INSERT INTO customers(customer_first_name, customer_last_name)
            VALUES (p_customer_first_name, p_customer_last_name) RETURNING  customer_id into p_customer_id;
            EXCEPTION WHEN unique_violation THEN
            -- 注：原语句存在拼写错误custmomer_id
            p_customer_id  = (SELECT custmomer_id 
                              FROM  customers
                              WHERE customer_first_name = p_customer_first_name AND
                                    customer_last_name = p_customer_last_name
        END;
    end if;    
   
    p_current_item_value := (select property_value
                             from customer_informations
                             where customer_id = p_customer_id AND
                                   customer_property_name = p_customer_property_name);
  

   
    if (p_current_item_value is NULL) THEN
        INSERT INTO customer_informations(customer_id, customer_property_name, customer_property_value)
        VALUES (p_customer_id, p_customer_property_name, p_customer_property_value);
    -- 注：原语句存在未定义变量p_item_value，应为p_customer_property_value
    elseif (p_current_item_value is not null AND p_current_item_value != p_item_value) then
        -- 注：原语句WHERE条件缺失，会更新该客户所有属性
        UPDATE customer_informations 
        SET customer_property_value = p_current_item_value
        WHERE  customer_id = p_customer_id ;        
    end if;
end; $procedure$;

表约束 #

CustomerInformations表 #

CONSTRAINT ux_customer_informations UNIQUE (customer_id, customer_property_name)

Customers表 #

CONSTRAINT ux_customers UNIQUE (customer_firstname, customer_lastname)

已尝试的优化 #

• 服务端并行处理：添加了存储过程的唯一违例捕获，但提速效果不明显
• 考虑删除唯一约束和索引：但担心重复数据清理难度（有其他人员操作数据库）

优化建议 #

1. 存储过程逻辑修正与优化 #

• 先修复代码错误：修正存储过程中的列数不匹配、拼写错误、未定义变量、WHERE条件缺失等问题，避免无效执行和数据错误。
• 用INSERT ... ON CONFLICT替代先查后写：
  针对Customers表，直接利用唯一约束简化客户创建逻辑，省去异常捕获的开销：

  INSERT INTO customers(customer_first_name, customer_last_name)
  VALUES (p_customer_first_name, p_customer_last_name)
  ON CONFLICT (customer_first_name, customer_last_name) DO NOTHING
  RETURNING customer_id INTO p_customer_id;
  
  IF p_customer_id IS NULL THEN
      SELECT customer_id INTO p_customer_id 
      FROM customers 
      WHERE customer_first_name = p_customer_first_name 
        AND customer_last_name = p_customer_last_name 
      LIMIT 1;
  END IF;
  

  针对CustomerInformations表，直接用ON CONFLICT DO UPDATE省去前置查询：

  INSERT INTO customer_informations(customer_id, customer_property_name, customer_property_value)
  VALUES (p_customer_id, p_customer_property_name, p_customer_property_value)
  ON CONFLICT (customer_id, customer_property_name) DO UPDATE
  SET customer_property_value = EXCLUDED.customer_property_value
  WHERE customer_informations.customer_property_value != EXCLUDED.customer_property_value;
  

2. 批量处理替代单条循环调用 #

• C#端分组批量处理：将相同姓名的客户属性分组，减少对Customers表的重复查询：

  var groupedData = request.Data.GroupBy(x => new { x.FirstName, x.LastName });
  foreach (var group in groupedData)
  {
      // 单次获取客户ID
      var customerId = await GetOrCreateCustomerAsync(group.Key.FirstName, group.Key.LastName, transaction);
      // 批量插入/更新该客户的所有属性
      var propertyParams = group.Select(x => new { customerId, x.PropertyName, x.PropertyValue });
      await db.ExecuteAsync(@"
          INSERT INTO customer_informations(customer_id, customer_property_name, customer_property_value)
          VALUES (@customerId, @PropertyName, @PropertyValue)
          ON CONFLICT (customer_id, customer_property_name) DO UPDATE
          SET customer_property_value = EXCLUDED.customer_property_value
          WHERE customer_informations.customer_property_value != EXCLUDED.customer_property_value;
      ", propertyParams, transaction: transaction);
  }
  

• 利用Dapper的批量执行能力，减少单条调用的网络往返开销。

3. 数据库层面优化 #

• 索引维护：低峰期执行REINDEX INDEX ux_customer_informations;清理索引碎片，同时调大maintenance_work_mem参数提升索引维护效率。
• 写入参数调优：临时调大wal_buffers和checkpoint_timeout减少WAL写入频率（操作前需做好备份，规避断电风险）。
• 慢查询分析：开启pg_stat_statements插件，查看存储过程的执行计划，定位具体慢查询节点。
• 分区表考虑：若允许调整表结构，可对CustomerInformations表按customer_id或时间范围做分区，降低单表数据量带来的IO压力。

4. 并行处理优化 #

• 限制并行度：根据数据库连接池大小设置合理的并行任务数（例如取连接池最大数的70%），避免连接耗尽导致阻塞。
• 批量并行：将分组后的数据拆分为多个批次，并行处理批次而非单条数据，平衡资源利用。

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