10 億行數據集處理挑戰：從 15 分鐘到 5 秒
作者：DeepNoMind
Golang 在處理 10 億行數據集的挑戰中展現了高效的并發處理和優化的 I/O 操作能力，通過使用 Parquet 二進制格式，進一步提升了數據處理性能，并最終將處理時間從 15 分鐘優化到了 5 秒。
0 億行挑戰[2]在編程界引起了廣泛關注，其主要目的是測試不同編程語言處理并匯總包含 10 億行的海量數據集的速度，而以性能和并發能力著稱的 Go 是有力競爭者。目前性能最好的 Java 實現處理數據的時間僅為 1.535 秒，我們看看 Go 的表現如何。
本文將基于 Go 特有的功能進行優化。注：所有基準數據都是在多次運行后計算得出的。
硬件設置
在配備 M2 Pro 芯片的 16 英寸 MacBook Pro 上進行了所有測試，該芯片擁有 12 個 CPU 核和 36 GB 內存。不同環境運行的測試結果可能因硬件而異，但相對性能差異應該差不多。
什么是 "10 億行挑戰"？
挑戰很簡單：處理包含 10 億個任意溫度測量值的文本文件，并計算每個站點的匯總統計數據（最小值、平均值和最大值）。問題在于如何高效處理如此龐大的數據集。
數據集通過代碼倉庫中的createMeasurements.go 腳本生成。運行腳本后，將得到一個 12.8 GB 大的由分號分隔的文本文件（measurements.txt），包含兩列數據：站點名稱和測量值。
我們需要處理該文件，并輸出以下結果：
{Station1=12.3/25.6/38.9, Station2=10.0/22.5/35.0, ...}我們看看如何實現這一目標。
Go 初始實現
1.單核版本
我們從基本的單線程實現開始。該版本逐行讀取文件，解析數據，并更新映射以跟蹤統計數據。
package main
import (
"bufio"
"fmt"
"os"
"strconv"
"strings"
)
type Stats struct {
Min float64
Max float64
Sum float64
Count int64
}
func processFile(filename string) {
file, err := os.Open(filename)
if err != nil {
panic(err)
}
defer file.Close()
statsMap := make(map[string]*Stats)
scanner := bufio.NewScanner(file)
for scanner.Scan() {
line := scanner.Text()
parts := strings.Split(line, ";")
if len(parts) != 2 {
continue
}
station := parts[0]
measurement, err := strconv.ParseFloat(parts[1], 64)
if err != nil {
continue
}
stat, exists := statsMap[station]
if !exists {
statsMap[station] = &Stats{
Min: measurement,
Max: measurement,
Sum: measurement,
Count: 1,
}
} else {
if measurement < stat.Min {
stat.Min = measurement
}
if measurement > stat.Max {
stat.Max = measurement
}
stat.Sum += measurement
stat.Count++
}
}
if err := scanner.Err(); err != nil {
panic(err)
}
fmt.Print("{")
for station, stat := range statsMap {
mean := stat.Sum / float64(stat.Count)
fmt.Printf("%s=%.1f/%.1f/%.1f, ", station, stat.Min, mean, stat.Max)
}
fmt.Print("\b\b} \n")
}
func main() {
processFile("data/measurements.txt")
}2.多核版本
為了充分利用多個 CPU 核，我們把文件分成若干塊，然后利用Goroutine 和Channel 并行處理。
package main
import (
"bufio"
"fmt"
"os"
"runtime"
"strconv"
"strings"
"sync"
)
type Stats struct {
Min float64
Max float64
Sum float64
Count int64
}
func worker(lines []string, wg *sync.WaitGroup, statsChan chan map[string]*Stats) {
defer wg.Done()
statsMap := make(map[string]*Stats)
for _, line := range lines {
parts := strings.Split(line, ";")
if len(parts) != 2 {
continue
}
station := parts[0]
measurement, err := strconv.ParseFloat(parts[1], 64)
if err != nil {
continue
}
stat, exists := statsMap[station]
if !exists {
statsMap[station] = &Stats{
Min: measurement,
Max: measurement,
Sum: measurement,
Count: 1,
}
} else {
if measurement < stat.Min {
stat.Min = measurement
}
if measurement > stat.Max {
stat.Max = measurement
}
stat.Sum += measurement
stat.Count++
}
}
statsChan <- statsMap
}
func processFile(filename string) {
file, err := os.Open(filename)
if err != nil {
panic(err)
}
defer file.Close()
numCPU := runtime.NumCPU()
linesPerWorker := 1000000
scanner := bufio.NewScanner(file)
lines := make([]string, 0, linesPerWorker)
var wg sync.WaitGroup
statsChan := make(chan map[string]*Stats, numCPU)
go func() {
wg.Wait()
close(statsChan)
}()
for scanner.Scan() {
lines = append(lines, scanner.Text())
if len(lines) >= linesPerWorker {
wg.Add(1)
go worker(lines, &wg, statsChan)
lines = make([]string, 0, linesPerWorker)
}
}
if len(lines) > 0 {
wg.Add(1)
go worker(lines, &wg, statsChan)
}
if err := scanner.Err(); err != nil {
panic(err)
}
finalStats := make(map[string]*Stats)
for partialStats := range statsChan {
for station, stat := range partialStats {
existingStat, exists := finalStats[station]
if !exists {
finalStats[station] = stat
} else {
if stat.Min < existingStat.Min {
existingStat.Min = stat.Min
}
if stat.Max > existingStat.Max {
existingStat.Max = stat.Max
}
existingStat.Sum += stat.Sum
existingStat.Count += stat.Count
}
}
}
fmt.Print("{")
for station, stat := range finalStats {
mean := stat.Sum / float64(stat.Count)
fmt.Printf("%s=%.1f/%.1f/%.1f, ", station, stat.Min, mean, stat.Max)
}
fmt.Print("\b\b} \n")
}
func main() {
processFile("data/measurements.txt")
}3.Go 實現結果
單核和多核版本的運行結果分別如下：
- 單核版本：15 分 30 秒
- 多核版本：6 分 45 秒
雖然多核版本有明顯改善，但處理數據仍然花了好幾分鐘。下面看看如何進一步優化。
利用 Go 的并發和緩沖 I/O 進行優化
為了提高性能，我們考慮利用緩沖 I/O，并優化 Goroutine。
package main
import (
"bufio"
"fmt"
"os"
"runtime"
"strconv"
"strings"
"sync"
)
type Stats struct {
Min float64
Max float64
Sum float64
Count int64
}
func worker(id int, jobs <-chan []string, results chan<- map[string]*Stats, wg *sync.WaitGroup) {
defer wg.Done()
for lines := range jobs {
statsMap := make(map[string]*Stats)
for _, line := range lines {
parts := strings.Split(line, ";")
if len(parts) != 2 {
continue
}
station := parts[0]
measurement, err := strconv.ParseFloat(parts[1], 64)
if err != nil {
continue
}
stat, exists := statsMap[station]
if !exists {
statsMap[station] = &Stats{
Min: measurement,
Max: measurement,
Sum: measurement,
Count: 1,
}
} else {
if measurement < stat.Min {
stat.Min = measurement
}
if measurement > stat.Max {
stat.Max = measurement
}
stat.Sum += measurement
stat.Count++
}
}
results <- statsMap
}
}
func processFile(filename string) {
file, err := os.Open(filename)
if err != nil {
panic(err)
}
defer file.Close()
numCPU := runtime.NumCPU()
jobs := make(chan []string, numCPU)
results := make(chan map[string]*Stats, numCPU)
var wg sync.WaitGroup
for i := 0; i < numCPU; i++ {
wg.Add(1)
go worker(i, jobs, results, &wg)
}
go func() {
wg.Wait()
close(results)
}()
scanner := bufio.NewScanner(file)
bufferSize := 1000000
lines := make([]string, 0, bufferSize)
for scanner.Scan() {
lines = append(lines, scanner.Text())
if len(lines) >= bufferSize {
jobs <- lines
lines = make([]string, 0, bufferSize)
}
}
if len(lines) > 0 {
jobs <- lines
}
close(jobs)
if err := scanner.Err(); err != nil {
panic(err)
}
finalStats := make(map[string]*Stats)
for partialStats := range results {
for station, stat := range partialStats {
existingStat, exists := finalStats[station]
if !exists {
finalStats[station] = stat
} else {
if stat.Min < existingStat.Min {
existingStat.Min = stat.Min
}
if stat.Max > existingStat.Max {
existingStat.Max = stat.Max
}
existingStat.Sum += stat.Sum
existingStat.Count += stat.Count
}
}
}
fmt.Print("{")
for station, stat := range finalStats {
mean := stat.Sum / float64(stat.Count)
fmt.Printf("%s=%.1f/%.1f/%.1f, ", station, stat.Min, mean, stat.Max)
}
fmt.Print("\b\b} \n")
}
func main() {
processFile("data/measurements.txt")
}優化 Go 實現結果
優化后，處理時間大大縮短：
多核優化版：3 分 50 秒
確實獲得了實質性改進，但仍然無法與最快的 Java 實現相媲美。
利用高效數據格式
由于文本文件不是處理大型數據集的最有效方式，我們考慮將數據轉換為 Parquet 等二進制格式來提高效率。
1.轉換為 Parquet
可以用 Apache Arrow 等工具將文本文件轉換為 Parquet 文件。為簡單起見，假設數據已轉換為measurements.parquet。
2.用 Go 處理 Parquet 文件
我們用parquet-go 庫來讀取 Parquet 文件。
package main
import (
"fmt"
"log"
"sort"
"github.com/xitongsys/parquet-go/reader"
"github.com/xitongsys/parquet-go/source/local"
)
type Measurement struct {
StationName string `parquet:"name=station_name, type=BYTE_ARRAY, convertedtype=UTF8, encoding=PLAIN_DICTIONARY"`
Measurement float64 `parquet:"name=measurement, type=DOUBLE"`
}
type Stats struct {
Min float64
Max float64
Sum float64
Count int64
}
func processParquetFile(filename string) {
fr, err := local.NewLocalFileReader(filename)
if err != nil {
log.Fatal(err)
}
defer fr.Close()
pr, err := reader.NewParquetReader(fr, new(Measurement), 4)
if err != nil {
log.Fatal(err)
}
defer pr.ReadStop()
num := int(pr.GetNumRows())
statsMap := make(map[string]*Stats)
for i := 0; i < num; i += 1000000 {
readNum := 1000000
if i+readNum > num {
readNum = num - i
}
measurements := make([]Measurement, readNum)
if err = pr.Read(&measurements); err != nil {
log.Fatal(err)
}
for _, m := range measurements {
stat, exists := statsMap[m.StationName]
if !exists {
statsMap[m.StationName] = &Stats{
Min: m.Measurement,
Max: m.Measurement,
Sum: m.Measurement,
Count: 1,
}
} else {
if m.Measurement < stat.Min {
stat.Min = m.Measurement
}
if m.Measurement > stat.Max {
stat.Max = m.Measurement
}
stat.Sum += m.Measurement
stat.Count++
}
}
}
stationNames := make([]string, 0, len(statsMap))
for station := range statsMap {
stationNames = append(stationNames, station)
}
sort.Strings(stationNames)
fmt.Print("{")
for _, station := range stationNames {
stat := statsMap[station]
mean := stat.Sum / float64(stat.Count)
fmt.Printf("%s=%.1f/%.1f/%.1f, ", station, stat.Min, mean, stat.Max)
}
fmt.Print("\b\b} \n")
}
func main() {
processParquetFile("data/measurements.parquet")
}3.Parquet 處理結果
通過以 Parquet 格式處理數據，取得了顯著的性能提升：
Parquet 處理時間：5 秒
Go 的性能更進一步接近了最快的 Java 實現。
結論
Go 在 10 億行挑戰中表現出色。通過利用 Go 的并發模型和優化 I/O 操作，大大縮短了處理時間。通過將數據集轉換為二進制格式（如 Parquet）可進一步提高性能。
主要收獲：
- Go 高效的并發機制使其適合處理大型數據集。
- 優化 I/O 和使用緩沖讀取可大幅提高性能。
- 利用 Parquet 等高效數據格式可大大縮短處理時間。
最終想法：
盡管 Go 可能無法取代速度最快的 Java 實現，但在高效處理大數據方面展示了令人印象深刻的能力。工具的選擇和優化可以縮小性能差距，使 Go 成為數據密集型任務的可行選擇。
參考資料
- [1]Go One Billion Row Challenge — From 15 Minutes to 5 Seconds:https://medium.com/@code-geass/go-one-billion-row-challenge-from-15-minutes-to-5-seconds-a1206611e230
- [2]10 億行挑戰:https://1brc.dev
責任編輯：趙寧寧
來源：
DeepNoMind
