我正在嘗試使用 Golang 使用 goroutines 加載一個大的 CSV 文件。csv 的維度是 (254882, 100)。但是當我解析 csv 并將其存儲到 2D 列表中時使用我的 goroutines，我得到的行小于 254882 并且每次運行的數量都不同。我覺得這是由于 goroutines 而發生的，但似乎無法指出原因。誰能幫幫我嗎。我也是 Golang 的新手。下面是我的代碼func loadCSV(csvFile string) (*[][]float64, error) { startTime := time.Now() var dataset [][]float64 f, err := os.Open(csvFile) if err != nil { return &dataset, err } r := csv.NewReader(bufio.NewReader(f)) counter := 0 var wg sync.WaitGroup for { record, err := r.Read() if err == io.EOF { break } if counter != 0 { wg.Add(1) go func(r []string, dataset *[][]float64) { var temp []float64 for _, each := range record { f, err := strconv.ParseFloat(each, 64) if err == nil { temp = append(temp, f) } } *dataset = append(*dataset, temp) wg.Done() }(record, &dataset) } counter++ } wg.Wait() duration := time.Now().Sub(startTime) log.Printf("Loaded %d rows in %v seconds", counter, duration) return &dataset, nil}我的主要功能如下所示func main() { // runtime.GOMAXPROCS(4) dataset, err := loadCSV("AvgW2V_train.csv") if err != nil { panic(err) } fmt.Println(len(*dataset))}如果有人也需要下載 CSV，請單擊下面的鏈接 (485 MB) https://drive.google.com/file/d/1G4Nw6JyeC-i0R1exWp5BtRtGM1Fwyelm/view?usp=sharing
2 回答
鴻蒙傳說
TA貢獻1865條經驗 獲得超7個贊
Go數據競爭檢測器
您的結果未定義，因為您有數據競爭。
~/gopath/src$ go run -race racer.go
==================
WARNING: DATA RACE
Write at 0x00c00008a060 by goroutine 6:
? runtime.mapassign_faststr()
? ? ? /home/peter/go/src/runtime/map_faststr.go:202 +0x0
? main.main.func2()
? ? ? /home/peter/gopath/src/racer.go:16 +0x6a
Previous write at 0x00c00008a060 by goroutine 5:
? runtime.mapassign_faststr()
? ? ? /home/peter/go/src/runtime/map_faststr.go:202 +0x0
? main.main.func1()
? ? ? /home/peter/gopath/src/racer.go:11 +0x6a
Goroutine 6 (running) created at:
? main.main()
? ? ? /home/peter/gopath/src/racer.go:14 +0x88
Goroutine 5 (running) created at:
? main.main()
? ? ? /home/peter/gopath/src/racer.go:9 +0x5b
==================
fatal error: concurrent map writes
==================
WARNING: DATA RACE
Write at 0x00c00009a088 by goroutine 6:
? main.main.func2()
? ? ? /home/peter/gopath/src/racer.go:16 +0x7f
Previous write at 0x00c00009a088 by goroutine 5:
? main.main.func1()
? ? ? /home/peter/gopath/src/racer.go:11 +0x7f
Goroutine 6 (running) created at:
? main.main()
? ? ? /home/peter/gopath/src/racer.go:14 +0x88
Goroutine 5 (running) created at:
? main.main()
? ? ? /home/peter/gopath/src/racer.go:9 +0x5b
==================
goroutine 34 [running]:
runtime.throw(0x49e156, 0x15)
? ? /home/peter/go/src/runtime/panic.go:608 +0x72 fp=0xc000094718 sp=0xc0000946e8 pc=0x44b342
runtime.mapassign_faststr(0x48ace0, 0xc00008a060, 0x49c9c3, 0x8, 0xc00009a088)
? ? /home/peter/go/src/runtime/map_faststr.go:211 +0x46c fp=0xc000094790 sp=0xc000094718 pc=0x43598c
main.main.func1(0x49c9c3, 0x8)
? ? /home/peter/gopath/src/racer.go:11 +0x6b fp=0xc0000947d0 sp=0xc000094790 pc=0x47ac6b
runtime.goexit()
? ? /home/peter/go/src/runtime/asm_amd64.s:1340 +0x1 fp=0xc0000947d8 sp=0xc0000947d0 pc=0x473061
created by main.main
? ? /home/peter/gopath/src/racer.go:9 +0x5c
goroutine 1 [sleep]:
time.Sleep(0x5f5e100)
? ? /home/peter/go/src/runtime/time.go:105 +0x14a
main.main()
? ? /home/peter/gopath/src/racer.go:19 +0x96
goroutine 35 [runnable]:
main.main.func2(0x49c9c3, 0x8)
? ? /home/peter/gopath/src/racer.go:16 +0x6b
created by main.main
? ? /home/peter/gopath/src/racer.go:14 +0x89
exit status 2
~/gopath/src$?
racer.go:
package main
import (
? ? "bufio"
? ? "encoding/csv"
? ? "fmt"
? ? "io"
? ? "log"
? ? "os"
? ? "strconv"
? ? "sync"
? ? "time"
)
func loadCSV(csvFile string) (*[][]float64, error) {
? ? startTime := time.Now()
? ? var dataset [][]float64
? ? f, err := os.Open(csvFile)
? ? if err != nil {
? ? ? ? return &dataset, err
? ? }
? ? r := csv.NewReader(bufio.NewReader(f))
? ? counter := 0
? ? var wg sync.WaitGroup
? ? for {
? ? ? ? record, err := r.Read()
? ? ? ? if err == io.EOF {
? ? ? ? ? ? break
? ? ? ? }
? ? ? ? if counter != 0 {
? ? ? ? ? ? wg.Add(1)
? ? ? ? ? ? go func(r []string, dataset *[][]float64) {
? ? ? ? ? ? ? ? var temp []float64
? ? ? ? ? ? ? ? for _, each := range record {
? ? ? ? ? ? ? ? ? ? f, err := strconv.ParseFloat(each, 64)
? ? ? ? ? ? ? ? ? ? if err == nil {
? ? ? ? ? ? ? ? ? ? ? ? temp = append(temp, f)
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? *dataset = append(*dataset, temp)
? ? ? ? ? ? ? ? wg.Done()
? ? ? ? ? ? }(record, &dataset)
? ? ? ? }
? ? ? ? counter++
? ? }
? ? wg.Wait()
? ? duration := time.Now().Sub(startTime)
? ? log.Printf("Loaded %d rows in %v seconds", counter, duration)
? ? return &dataset, nil
}
func main() {
? ? // runtime.GOMAXPROCS(4)
? ? dataset, err := loadCSV("/home/peter/AvgW2V_train.csv")
? ? if err != nil {
? ? ? ? panic(err)
? ? }
? ? fmt.Println(len(*dataset))
}
狐的傳說
TA貢獻1804條經驗 獲得超3個贊
沒有必要使用，*[][]float64因為那將是一個雙指針。
我對你的程序做了一些小的修改。
dataset可用于新的 goroutine，因為它是在它上面的代碼塊中聲明的。
similarlyrecord也是可用的，但是由于recordvariable 是不時變化的，我們需要將它傳遞給新的 goroutine。
雖然不需要傳遞dataset，因為它沒有改變，而這正是我們想要的，這樣我們就可以將 temp 附加到dataset.
但是當多個 goroutines 試圖附加到同一個變量時，就會發生競爭條件，即多個 goroutines 試圖寫入同一個變量。
所以我們需要確保在任何時候只有一個 can goroutine 可以添加。所以我們使用鎖來進行順序追加。
package main
import (
"bufio"
"encoding/csv"
"fmt"
"os"
"strconv"
"sync"
)
func loadCSV(csvFile string) [][]float64 {
var dataset [][]float64
f, _ := os.Open(csvFile)
r := csv.NewReader(f)
var wg sync.WaitGroup
l := new(sync.Mutex) // lock
for record, err := r.Read(); err == nil; record, err = r.Read() {
wg.Add(1)
go func(record []string) {
defer wg.Done()
var temp []float64
for _, each := range record {
if f, err := strconv.ParseFloat(each, 64); err == nil {
temp = append(temp, f)
}
}
l.Lock() // lock before writing
dataset = append(dataset, temp) // write
l.Unlock() // unlock
}(record)
}
wg.Wait()
return dataset
}
func main() {
dataset := loadCSV("train.csv")
fmt.Println(len(dataset))
}
有些錯誤沒有得到處理以使其最小化，但您應該處理錯誤。
- 2 回答
- 0 關注
- 179 瀏覽
添加回答
舉報
0/150
提交
取消