10 億行數據集處理挑戰：從 15 分鐘到 5 秒

作者：DeepNoMind 2024-12-19 07:50:00

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

Golang 數據集