如何在Go中调用Python Pandas脚本并发处理CSV文件？

你的思路是否可行？ #

完全可行，但要注意两个核心问题：Python进程的启动开销和Go与Python之间的数据传递效率。每个goroutine启动新Python进程会带来一定的资源消耗，同时数据从Go内存传递到Python进程需要选择合适的方式。

具体实现方案 #

方案1：每个goroutine独立启动Python进程（最简单直接） #

直接用Go的os/exec包启动Python解释器，把已加载的CSV内容通过标准输入(stdin)传给Python脚本，处理后再通过标准输出(stdout)返回结果。

Go代码示例：

package main

import (
    "bytes"
    "log"
    "os/exec"
    "sync"
)

// 单个CSV的处理逻辑
func processSingleCSV(csvData []byte, wg *sync.WaitGroup) {
    defer wg.Done()

    // Python处理脚本：从stdin读CSV，用Pandas处理后输出结果
    pyScript := `
import sys
import pandas as pd

# 读取stdin中的CSV数据
df = pd.read_csv(sys.stdin)
# 这里替换成你的实际处理逻辑（比如清洗、统计）
processed_df = df[df["value"] > 100]
# 将处理结果输出到stdout
print(processed_df.to_csv(index=False))
`

    cmd := exec.Command("python3", "-c", pyScript)
    cmd.Stdin = bytes.NewReader(csvData)
    
    var output bytes.Buffer
    cmd.Stdout = &output
    
    if err := cmd.Run(); err != nil {
        log.Printf("处理CSV失败: %v", err)
        return
    }

    // 这里可以处理Python返回的结果
    log.Println("处理完成，结果长度:", len(output.Bytes()))
}

func main() {
    // 假设loadedCSVs是你已经加载好的CSV字节数组切片
    loadedCSVs := [][]byte{/* 你的CSV数据 */}
    
    var wg sync.WaitGroup
    // 控制并发数：避免同时启动过多Python进程耗尽资源
    sem := make(chan struct{}, 4) // 限制最多4个并发进程

    for _, csv := range loadedCSVs {
        wg.Add(1)
        sem <- struct{}{}
        go func(data []byte) {
            defer func() {
                wg.Done()
                <-sem
            }()
            processSingleCSV(data, &wg)
        }(csv)
    }

    wg.Wait()
}

注意事项：

• 用**信号量(semaphore)**控制并发进程数，避免系统资源被占满。
• 数据传递用stdin/stdout比临时文件更高效，减少磁盘IO开销。

方案2：复用Python进程（适合大量CSV场景） #

如果CSV数量极多，反复启动Python进程的开销会很明显。可以提前启动一批Python worker进程，Go通过管道与它们通信，循环分发CSV数据处理。

核心思路：

1. 启动N个Python进程，每个进程持续监听stdin，处理完一个CSV后等待下一个任务。
2. Go通过通道管理这些worker，把CSV数据分发给空闲的worker，接收处理结果。

Python Worker脚本示例（worker.py）：

import sys
import pandas as pd
from io import StringIO

def process_csv():
    while True:
        # 读取完整的CSV数据（可自定义分隔符或长度前缀实现边界识别）
        csv_content = sys.stdin.read()
        if not csv_content:
            break
        df = pd.read_csv(StringIO(csv_content))
        # 处理逻辑...
        result = df.describe().to_csv()
        print(result)
        sys.stdout.flush() # 确保结果立即输出

if __name__ == "__main__":
    process_csv()

Go侧负责启动worker、分发数据和收集结果，这种方式能大幅减少进程启动的开销。

方案3：使用Go-Python绑定库（不推荐生产环境） #

有一些CGo绑定库可以让Go直接调用Python函数（如go-python3、go-python），但存在以下问题：

• Python的GIL会限制并发，同一时间只能有一个线程执行Python代码，并发处理效率低。
• CGo带来跨平台编译复杂度，且容易出现内存泄漏问题。
• 维护状态不稳定，部分库可能停止更新。

除非你对性能要求不高且愿意处理这些问题，否则不建议采用。

方案选择建议 #

• 若CSV数量较少（几十到上百个）：选方案1，实现简单，无需额外依赖。
• 若CSV数量巨大（上千个以上）：选方案2，复用进程降低开销。
• 避免使用Go-Python绑定库，除非有特殊场景需求。

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