如何使用pgx.CopyFrom批量插入CSV数据到PostgreSQL数据库

方案可行性 #

你的实现思路完全可行，pgx.CopyFrom 底层对接PostgreSQL原生COPY协议，是pgx提供的大批量数据导入性能最优的接口，导入效率远高于普通INSERT批量写入，非常适合CSV全量导入场景。

类型不匹配问题原因 #

你遇到的类型错误来自两个问题：

• pgx.CopyFromRows 接收的是多行数据集合，类型为[][]interface{}，即第一层切片的每个元素代表一行数据，你直接传入单行的[]string类型line，维度不匹配。
• 单行数据要求是[]interface{}类型，不能直接传入[]string，需要做显式类型转换。

正确实现代码 #

注意不要逐行调用CopyFrom，否则完全无法发挥COPY协议的性能优势，需要按固定批次攒够数据后批量提交，参考实现如下：

const (
    timeFormat = "你的时间格式" // 替换为实际使用的时间格式
    batchSize  = 1000         // 批次大小可根据实际场景调整，建议取值1000-10000
)

// 初始化行缓存，预分配容量减少内存拷贝
rows := make([][]interface{}, 0, batchSize)

for {
    line, err := csvReader.Read()
    if err == io.EOF {
        // 文件读取结束，提交剩余缓存的行
        if len(rows) > 0 {
            _, copyErr := db.CopyFrom(
                context.Background(),
                // 注意：Identifier需要按标识符分段传入，不要把带点的全路径写在单个字符串里
                // 三段式为 数据库名.Schema名.表名，连接已指定数据库时可省略数据库名
                pgx.Identifier{"emms", "scada_crwf", "pwr_active"},
                col_headings,
                pgx.CopyFromRows(rows),
            )
            if copyErr != nil {
                log.Error("Final batch copy failed", "Loading Loop", copyErr)
            }
        }
        break
    }
    if err != nil {
        log.Error("Error reading csv data", "Loading Loop", err)
        continue
    }

    // 校验时间戳，跳过非数据行
    parseTime, err := time.Parse(timeFormat, line[0])
    if err != nil {
        continue
    }

    // 将单行[]string转换为[]interface{}
    row := make([]interface{}, len(line))
    for i, val := range line {
        // 建议对非字符串列做显式类型转换，避免隐式转换开销和格式错误
        // 比如第一列是timestamp类型，直接传入解析好的parseTime，不要传原始字符串
        // 数值、布尔类型同理，提前转成对应Go类型
        if i == 0 {
            row[i] = parseTime
            continue
        }
        row[i] = val
    }
    rows = append(rows, row)

    // 攒够批次大小就执行一次CopyFrom
    if len(rows) == batchSize {
        _, err = db.CopyFrom(
            context.Background(),
            pgx.Identifier{"emms", "scada_crwf", "pwr_active"},
            col_headings,
            pgx.CopyFromRows(rows),
        )
        if err != nil {
            log.Error("Batch copy failed", "Loading Loop", err)
        }
        // 重置缓存，复用已分配的内存
        rows = rows[:0]
    }
}

关键注意事项 #

• pgx.Identifier 传参错误是高频坑：PostgreSQL中用点分隔的多级标识符需要拆成单独的字符串传入，不要把emms.scada_crwf.pwr_active整体作为一个字符串，否则会被识别为名称带点的单级表，触发「表不存在」错误。
• 必须攒批提交：逐行调用CopyFrom会产生大量网络交互，性能甚至不如普通INSERT，按1000~10000行的批次提交才能发挥COPY协议的性能优势。
• 提前做类型转换：对于表中时间、数值、布尔类型的列，不要直接传入原始CSV字符串，提前转换为对应的Go类型（time.Time、int64、float64、bool），既可以减少数据库侧隐式转换的开销，也能提前过滤格式错误的脏数据。
• 批次大小按需调整：如果导入机器内存充足，可以把批次大小上调到5000~10000进一步提升速度；如果内存紧张，适当下调到500左右即可。

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