如何用Python高效更新Postgres大表中的某一列

针对1200万行大表逐行更新效率极低的问题，以下是几个能大幅提升速度的实操方案：

• 批量更新（用psycopg2的execute_values）
  把多条更新请求打包成一个数据库查询，减少Python和Postgres之间的网络往返次数，这是最容易实现的优化方式。
  示例代码：

  from psycopg2.extras import execute_values
  import psycopg2
  
  # 建立连接
  conn = psycopg2.connect("dbname=your_db user=your_user password=your_pass host=your_host")
  cursor = conn.cursor()
  
  # 假设data是包含(主键key, 新列值)的列表，比如[(1001, "new_val1"), (1002, "new_val2"), ...]
  update_sql = """
  UPDATE db.table t
  SET col = data.new_col
  FROM (VALUES %s) AS data(key_val, new_col)
  WHERE t.key = data.key_val
  """
  # 批量大小建议设为1000-10000，根据内存和网络带宽调整
  execute_values(cursor, update_sql, data, page_size=5000)
  conn.commit()
  
  cursor.close()
  conn.close()
  

  这种方式能把更新速度提升几十到上百倍，完全避开逐行循环的低效问题。

• COPY导入临时表+批量更新
  对于超大规模数据，先把要更新的数据用Postgres的COPY命令导入临时表，再通过表关联完成更新——这是Postgres处理海量数据更新的最优方式之一，网络开销极小。
  步骤示例：

  from io import StringIO
  import psycopg2
  
  conn = psycopg2.connect("dbname=your_db user=your_user password=your_pass host=your_host")
  cursor = conn.cursor()
  
  # 1. 创建临时表（会话结束自动删除）
  cursor.execute("""
  CREATE TEMP TABLE temp_updates (
      key_val INT PRIMARY KEY,  -- 类型要和原表的key字段一致
      new_col VARCHAR(255)     -- 类型要和原表的col字段一致
  ) ON COMMIT DROP;
  """)
  
  # 2. 把数据转换成CSV格式，用COPY快速导入临时表
  data_stream = StringIO()
  for key, val in your_update_data:  # your_update_data是(key, new_col)的列表
      data_stream.write(f"{key},{val}\n")
  data_stream.seek(0)  # 回到流的起始位置
  
  cursor.copy_from(data_stream, 'temp_updates', columns=('key_val', 'new_col'), sep=',')
  
  # 3. 通过表关联批量更新原表
  cursor.execute("""
  UPDATE db.table t
  SET col = tu.new_col
  FROM temp_updates tu
  WHERE t.key = tu.key_val;
  """)
  conn.commit()
  
  cursor.close()
  conn.close()
  

  COPY是Postgres原生的高速数据导入工具，比批量INSERT还要快几倍，后续的表关联更新几乎是纯数据库内部操作，效率极高。

• 临时调整Postgres配置加速更新
  针对一次性的批量更新操作，可以临时修改会话级别的数据库参数，减少IO等待：

  cursor.execute("SET wal_buffers = '64MB';")  # 增大WAL缓冲区，减少磁盘写入次数
  cursor.execute("SET synchronous_commit = off;")  # 关闭同步提交，大幅提升写性能（注意：适合非核心数据，极端情况可能丢失最近的提交）
  conn.autocommit = False  # 关闭自动提交，减少事务开销
  

  注意：这些配置只适合本次批量更新，操作完成后建议改回默认值，避免影响其他业务的稳定性。

• 分批更新，避免锁表
  如果全表一次性更新会导致锁表影响线上业务，可以分批次处理，每次更新一部分数据：

  batch_size = 100000  # 每次更新10万行，可根据业务调整
  offset = 0
  key_to_val = dict(your_update_data)  # 把更新数据转成字典，方便快速查找
  
  while True:
      # 获取当前批次的主键列表（前提是key字段有索引，否则这个查询会很慢）
      cursor.execute("SELECT key FROM db.table LIMIT %s OFFSET %s;", (batch_size, offset))
      current_keys = [row[0] for row in cursor.fetchall()]
      if not current_keys:
          break
  
      # 构造当前批次的更新数据
      batch_data = [(key, key_to_val[key]) for key in current_keys]
      # 用execute_values批量更新
      execute_values(cursor, update_sql, batch_data, page_size=5000)
      conn.commit()
  
      offset += batch_size
  

  这种方式能降低单事务的锁持有时间，减少对业务的影响，同时保持较高的更新效率。

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