本文介绍如何通过 goroutine 并发模型与 bufio.Scanner 结合，优化大型 CSV 文件的 ETL 处理流程，包括提取、转换、加载三阶段的并行化设计及关键注意事项。
本文介绍如何通过 goroutine 并发模型与 `bufio.scanner` 结合，优化大型 csv 文件的 etl 处理流程，包括提取、转换、加载三阶段的并行化设计及关键注意事项。
在处理大型 CSV 文件（例如每行约 300 字节、总大小达 GB 级）时，串行读取—解析—入库极易成为性能瓶颈。Go 语言原生支持轻量级并发，合理使用 goroutine 可显著提升吞吐量。但需注意：并发 ≠ 并行，更不等于盲目增加 goroutine 数量。
推荐架构：ETL 三阶段流水线（Pipeline）
将 ETL 拆分为三个独立职责的 goroutine，通过有缓冲 channel 协作，形成高效流水线：
- Extract（E）：单 goroutine 顺序读取文件，逐行发送至 chan string（避免多 goroutine 竞争文件偏移）；
- Transform（T）：N 个 goroutine 并发消费原始行，执行 strings.Split()、字段校验、结构体映射、JSON 序列化等 CPU 密集操作；
- Load（L）：M 个 goroutine 异步写入数据库（如使用 sql.DB 的连接池），避免阻塞上游。
func runETLPipeline(filename string, workers int) error {
// 1. Extract: 单 goroutine 安全读取
lines := make(chan string, 1024)
go func() {
defer close(lines)
f, err := os.Open(filename)
if err != nil {
panic(err)
}
defer f.Close()
scanner := bufio.NewScanner(f)
for scanner.Scan() {
lines <- scanner.Text() // 避免传递 []byte 引用，防止内存逃逸
}
}()
// 2. Transform: 并发解析
jsonBytes := make(chan []byte, 1024)
for i := 0; i < workers; i++ {
go func() {
for line := range lines {
record := parseCSVLine(line) // 自定义解析逻辑
data, _ := json.Marshal(record)
jsonBytes <- data
}
}()
}
// 3. Load: 批量/异步写入 DB（示例伪代码）
var wg sync.WaitGroup
for i := 0; i < 2; i++ { // 通常 2–4 个 load goroutine 足够
wg.Add(1)
go func() {
defer wg.Done()
for data := range jsonBytes {
insertIntoDB(data) // 使用 prepared statement + connection pool
}
}()
}
wg.Wait()
return nil
}关于“从任意位置随机读取”的说明
bufio.Scanner 不支持随机偏移读取——它依赖底层 io.Reader 的顺序性，且内部维护缓冲区状态。若需分片并行读取（如多 goroutine 同时处理不同文件段），应：
QuickBooks Mileage CSV
下载
使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
- 使用 os.Seek() 定位到字节偏移；
- 手动按行切分：跳过不完整行（确保从 \n 后开始），再用 bufio.NewReader(io.LimitReader(...)) 构建局部扫描器；
- 更推荐方案：预处理生成行偏移索引（如记录每行起始位置），或改用 encoding/csv.Reader 配合 io.Seeker（需自行处理换行边界）。
⚠️ 注意事项：
- 避免无缓冲 channel 导致 goroutine 死锁；
- scanner.Text() 返回的字符串指向内部缓冲区，若需长期持有，请 string(scanner.Bytes()) 拷贝；
- 数据库写入务必复用连接池，禁用 defer db.Close() 在 goroutine 内；
- 监控 goroutine 泄漏：使用 runtime.NumGoroutine() 或 pprof。
总结：合理划分 ETL 阶段、控制 worker 数量（建议从 runtime.NumCPU() 开始调优）、配合缓冲 channel 与资源复用，即可在低内存开销下实现高吞吐 ETL。并发的本质是解耦阻塞点，而非堆砌 goroutine。
