如何用Go将Docker容器内CSV文件高效导入PostgreSQL？

为什么原始COPY命令失败 #

PostgreSQL的COPY <table> FROM '<path>'是让数据库服务器进程读取文件，路径必须是数据库服务器能访问的文件系统路径，不是你的Go程序所在Docker容器里的路径。如果数据库和Go程序不在同一个容器、也没共享存储卷，服务器自然找不到文件，这就是报错的核心原因。

最优方案1：使用COPY FROM STDIN（推荐） #

利用PostgreSQL的COPY ... FROM STDIN语法，让Go程序把CSV内容直接通过数据库连接流式传给服务器，性能比逐行插入高几个数量级，还能完整保留你的数据校验逻辑。

示例代码 #

import (
    "database/sql"
    "encoding/csv"
    "fmt"
    "io"
    "log"
    "os"
    "github.com/lib/pq" // 需要pq驱动支持COPY功能
)

func ImportCSVToPostgres(tableName, path string, fieldsQty int) error {
    file, err := os.Open(path)
    if err != nil {
        return fmt.Errorf("打开文件失败: %w", err)
    }
    defer file.Close()

    reader := csv.NewReader(file)
    // 先读取表头
    _, err := reader.Read()
    if err != nil {
        if err == io.EOF {
            return fmt.Errorf("文件为空")
        }
        return fmt.Errorf("读取表头失败: %w", err)
    }

    // 构建COPY命令，用QuoteIdentifier避免表名注入风险
    copyCmd := fmt.Sprintf("COPY %s FROM STDIN WITH (FORMAT CSV, HEADER false)", pq.QuoteIdentifier(tableName))
    tx, err := db.Pdb.Begin()
    if err != nil {
        return fmt.Errorf("开启事务失败: %w", err)
    }
    defer tx.Rollback()

    stmt, err := tx.Prepare(copyCmd)
    if err != nil {
        return fmt.Errorf("准备COPY语句失败: %w", err)
    }
    defer stmt.Close()

    rowCount := 0
    for {
        record, err := reader.Read()
        if err != nil {
            if err == io.EOF {
                break
            }
            return fmt.Errorf("读取CSV行失败: %w", err)
        }

        // 保留你的数据校验逻辑
        if len(record) != fieldsQty {
            return fmt.Errorf("第%d行字段数量错误: 预期%d个，实际%d个", rowCount+1, fieldsQty, len(record))
        }
        for idx, v := range record {
            if v == "" {
                return fmt.Errorf("第%d行第%d个字段为空", rowCount+1, idx+1)
            }
        }
        if _, err := uuid.Parse(record[0]); err != nil {
            return fmt.Errorf("第%d行ID格式错误: %s，需为UUID: %w", rowCount+1, record[0], err)
        }

        // 将当前行数据传给COPY进程
        _, err = stmt.Exec(record...)
        if err != nil {
            return fmt.Errorf("导入第%d行失败: %w", rowCount+1, err)
        }
        rowCount++
    }

    if rowCount == 0 {
        return fmt.Errorf("CSV文件仅含表头，无数据行")
    }

    // 提交事务完成导入
    if err := tx.Commit(); err != nil {
        return fmt.Errorf("提交事务失败: %w", err)
    }

    log.Printf("成功向%s导入%d行数据", tableName, rowCount)
    return nil
}

优势 #

• 性能碾压逐行INSERT：COPY是PostgreSQL原生的批量写入机制，大幅减少数据库交互开销
• 保留原有校验逻辑：提前拦截错误数据，避免脏数据入库
• 无需移动文件：直接在Go程序所在容器内完成处理，不用折腾文件共享

最优方案2：让数据库服务器能访问CSV文件 #

如果不想修改代码，可以解决原始COPY命令的路径问题：

• 挂载共享存储卷：把存放CSV的目录同时挂载到Go程序容器和PostgreSQL容器，比如启动容器时加-v /宿主机/csv目录:/容器内/csv目录，两个容器用同一路径，COPY命令直接用容器内路径即可
• 复制文件到数据库容器：用docker cp <本地CSV文件> <PostgreSQL容器名>:/tmp/把文件传到数据库容器，然后进入容器执行psql -U <用户名> -d <数据库名> -c "COPY <表名> FROM '/tmp/<文件名>' DELIMITER ',' CSV HEADER;"

现有代码的优化（若坚持用INSERT） #

如果因特殊原因不能用COPY，至少改成批量插入，减少数据库连接的交互次数：

import "strings"

// 示例：每1000行执行一次批量插入
func BatchInsertCSV(tableName, path string, fieldsQty int) error {
    // 省略文件打开、读取表头、校验逻辑...

    batchSize := 1000
    var values []interface{}
    baseSQL := ""

    // 根据表名构建批量INSERT的基础SQL
    switch tableName {
    case "<table_name1>":
        baseSQL = `INSERT INTO <table_name1> (field1, field2, field3, field4, field5) VALUES `
        // 生成批量占位符：($1,$2,$3,$4,$5),($6,$7,$8,$9,$10)...
        placeholders := make([]string, batchSize)
        for i := 0; i < batchSize; i++ {
            start := i*5 + 1
            placeholders[i] = fmt.Sprintf("($%d,$%d,$%d,$%d,$%d)", start, start+1, start+2, start+3, start+4)
        }
        baseSQL += strings.Join(placeholders, ",")
    case "<table_name2>":
        baseSQL = `INSERT INTO <table_name2> (field1, field2) VALUES `
        placeholders := make([]string, batchSize)
        for i := 0; i < batchSize; i++ {
            start := i*2 + 1
            placeholders[i] = fmt.Sprintf("($%d,$%d)", start, start+1)
        }
        baseSQL += strings.Join(placeholders, ",")
    // 处理table3...
    }

    rowCount := 0
    for {
        record, err := reader.Read()
        if err != nil {
            if err == io.EOF {
                break
            }
            return err
        }
        // 省略校验逻辑...

        values = append(values, record...)
        rowCount++

        // 达到批量阈值就执行插入
        if len(values) >= batchSize*fieldsQty {
            _, err = db.Pdb.Exec(baseSQL, values...)
            if err != nil {
                return fmt.Errorf("批量插入失败: %w", err)
            }
            values = nil // 重置批量数据
        }
    }

    // 处理剩余不足批量的行
    if len(values) > 0 {
        // 重新生成对应数量的占位符
        remaining := len(values)/fieldsQty
        placeholders := make([]string, remaining)
        switch tableName {
        case "<table_name1>":
            for i := 0; i < remaining; i++ {
                start := i*5 + 1
                placeholders[i] = fmt.Sprintf("($%d,$%d,$%d,$%d,$%d)", start, start+1, start+2, start+3, start+4)
            }
        case "<table_name2>":
            for i := 0; i < remaining; i++ {
                start := i*2 + 1
                placeholders[i] = fmt.Sprintf("($%d,$%d)", start, start+1)
            }
        }
        finalSQL := strings.Replace(baseSQL, strings.Join(make([]string, batchSize), ","), strings.Join(placeholders, ","), 1)
        _, err = db.Pdb.Exec(finalSQL, values...)
        if err != nil {
            return fmt.Errorf("插入剩余行失败: %w", err)
        }
    }

    log.Printf("成功插入%d行数据", rowCount)
    return nil
}

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