PostgreSQL+psycopg：调用ST_MakePoint批量插入海量点云数据方案咨询

针对数十亿条点云数据的批量插入需求，结合psycopg提供以下三种高效实现方案，均支持直接调用PostgreSQL端的ST_MakePoint函数：

方案1：使用execute_batch执行参数化批量插入 #

这是最直接的批量插入方式，通过psycopg的execute_batch将批量参数传递给包含ST_MakePoint的插入语句，避免单条插入的网络和解析开销。

import psycopg2
from psycopg2.extras import execute_batch

# 建立数据库连接
conn = psycopg2.connect(
    dbname="your_database",
    user="your_user",
    password="your_password",
    host="your_host"
)
cur = conn.cursor()

# 模拟批量点云数据，实际可从文件/数据流分批读取
batch_data = [
    (1, 32.656, 1.1, 2.2, 3.3),
    (1, 32.657, 4.4, 5.5, 6.6),
    # 更多数据项...
]

# 定义带参数占位符的插入语句，直接调用ST_MakePoint
insert_sql = """
INSERT INTO points_postgis (id_scan, scandist, pt)
VALUES (%s, %s, ST_MakePoint(%s, %s, %s))
"""

# 执行批量插入，batch_size根据内存情况调整（建议1万-10万条/批）
execute_batch(cur, insert_sql, batch_data, batch_size=10000)

conn.commit()
cur.close()
conn.close()

• 注意事项：
  • 根据本地内存容量调整batch_size，过大易引发内存溢出，过小则无法发挥批量优势。
  • 关闭自动提交，批量插入完成后一次性提交事务，减少磁盘IO开销。

方案2：使用copy_expert结合自定义COPY命令 #

利用PostgreSQL的COPY协议实现高速写入，通过copy_expert执行包含函数调用的自定义COPY语句，性能比execute_batch更优，适合超海量数据。

import psycopg2
from io import StringIO

conn = psycopg2.connect(
    dbname="your_database",
    user="your_user",
    password="your_password",
    host="your_host"
)
cur = conn.cursor()

# 准备内存缓冲区，将点云数据按文本格式写入（可分批写入避免内存过载）
data_buffer = StringIO()
for item in batch_data:
    # 用制表符分隔字段，和后续COPY命令的DELIMITER保持一致
    data_buffer.write(f"{item[0]}\t{item[1]}\t{item[2]}\t{item[3]}\t{item[4]}\n")
data_buffer.seek(0)  # 重置缓冲区指针到开头

# 定义带预处理逻辑的COPY命令，通过DO块调用ST_MakePoint
copy_sql = """
COPY points_postgis (id_scan, scandist, pt)
FROM STDIN WITH (FORMAT text, DELIMITER '\t')
AS (id_scan integer, scandist double precision, x double precision, y double precision, z double precision)
DO $$
BEGIN
    NEW.pt = ST_MakePoint(NEW.x, NEW.y, NEW.z);
END $$;
"""

# 执行COPY操作
cur.copy_expert(copy_sql, data_buffer)

conn.commit()
cur.close()
conn.close()

• 注意事项：
  • 若数据量极大无法一次性放入内存，可分块读取点云数据，分块写入缓冲区执行COPY。
  • 确保文本格式与COPY命令的分隔符、字段类型匹配，避免解析错误。

方案3：优化版临时表批量插入 #

针对你之前尝试临时表性能差的问题，通过配置无日志临时表、关闭约束来优化写入效率：

import psycopg2
from psycopg2.extras import execute_batch

conn = psycopg2.connect(...)
cur = conn.cursor()

# 创建无日志临时表，避免WAL日志写入，提升写入速度
cur.execute("""
CREATE UNLOGGED TEMP TABLE temp_points (
    id_scan integer,
    scandist double precision,
    x double precision,
    y double precision,
    z double precision
) ON COMMIT DROP;
""")

# 批量插入原始数据到临时表（也可替换为copy_expert进一步提升速度）
execute_batch(cur, "INSERT INTO temp_points VALUES (%s, %s, %s, %s, %s)", batch_data, batch_size=100000)

# 从临时表插入目标表，批量执行ST_MakePoint
cur.execute("""
INSERT INTO points_postgis (id_scan, scandist, pt)
SELECT id_scan, scandist, ST_MakePoint(x, y, z)
FROM temp_points;
""")

conn.commit()
cur.close()
conn.close()

• 注意事项：
  • ON COMMIT DROP确保临时表在事务结束后自动清理，无需手动删除。
  • 临时表无需创建索引和约束，插入完成后再通过SELECT语句转换写入目标表。

内容的提问来源于stack exchange，提问作者Rafael Scudelari de Macedo