Lösung der Milliarden-Zeilen-Herausforderung in Go (von bis s)

Vor einiger Zeit erzählte mir ein Freund von einer Herausforderung, bei der es darum ging, eine Datei mit 1 Milliarde Zeilen zu lesen. Ich fand die Idee sehr interessant, aber da es Prüfungswoche an der Uni war, habe ich sie aufgegeben, um sie mir später anzusehen. Monate später sah ich ein Video von Theo über die Herausforderung und beschloss, es mir genauer anzusehen.

Das Ziel der One Billion Row Challenge besteht darin, die minimale, maximale und durchschnittliche Temperatur einer Liste von Städten zu berechnen. Das Detail besteht darin, dass diese Liste 1 Milliarde Elemente enthält, wobei jedes Element aus dem Namen einer Stadt besteht und eine Temperatur. Jede Stadt kann mehr als einmal vorkommen. Schließlich muss das Programm diese Werte in alphabetischer Reihenfolge nach Städtenamen anzeigen.

Ich dachte, es würde Spaß machen, die Herausforderung zu lösen, auch wenn es keine Belohnung gab. Jedenfalls habe ich diesen Text geschrieben, der meinen Prozess beschreibt.

Erster Versuch: Den Code zum Laufen bringen

Immer wenn ich ein komplizierteres Problem lösen muss, besteht mein erstes Ziel darin, das Programm zum Laufen zu bringen. Es ist vielleicht nicht der schnellste oder sauberste Code, aber es ist Code, der funktioniert.

Im Grunde habe ich die Standortstruktur erstellt, um jede Stadt in der Liste darzustellen. Sie enthält die minimale und maximale Temperatur, die Summe der Temperaturen und wie oft die Stadt in der Liste erscheint (die letzten beiden sind zur Berechnung des Durchschnitts erforderlich). . Ich weiß, dass es eine Möglichkeit gibt, den Durchschnitt direkt zu berechnen, ohne die Anzahl der Temperaturen und deren Summe speichern zu müssen. Aber wie ich bereits erwähnt habe, bestand das Ziel darin, dass der Code funktioniert.

Die Datenliste besteht aus dem Namen der Stadt, gefolgt von der Temperatur, getrennt durch ein Semikolon. Zum Beispiel:

Antananarivo;15.6
Iqaluit;-20.7
Dolisie;25.8
Kuopio;-6.8

Der einfachste Weg, Daten zu lesen, ist die Verwendung von Scan, mit dem Sie jeweils eine Zeile lesen können. Mit der Zeile können Sie sie in zwei Teile teilen: die Werte vor und nach dem Semikolon. Um die Temperatur zu ermitteln, können Sie strconv.ParseFloat verwenden, das einen String in einen Float umwandelt. Der vollständige Code für die erste Implementierung ist unten zu sehen:

package main

import (
    "bufio"
    "fmt"
    "math"
    "os"
    "sort"
    "strconv"
    "strings"
)

type Location struct {
    min   float64
    max   float64
    sum   float64
    count int
}

func NewLocation() *Location {
    return &Location{
        min:   math.MaxInt16,
        max:   math.MinInt16,
        sum:   0,
        count: 0,
    }
}

func (loc *Location) Add(temp float64) {
    if temp < loc.min {
        loc.min = temp
    } else if temp > loc.max {
        loc.max = temp
    }

    loc.sum += temp
    loc.count += 1
}

var cpuprofile = flag.String("cpuprofile", "", "write cpu profile to file")

func main() {
    flag.Parse()
    if *cpuprofile != "" {
        f, err := os.Create(*cpuprofile)
        if err != nil {
            log.Fatal(err)
        }
        pprof.StartCPUProfile(f)
        defer pprof.StopCPUProfile()
    }

    file, _ := os.Open("./measurements.txt")
    defer file.Close()

    m := map[string]*Location{}

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        line := scanner.Text()
        name, tempStr, _ := strings.Cut(line, ";")
        temp, _ := strconv.ParseFloat(tempStr, 32)

        loc, ok := m[name]
        if !ok {
            loc = NewLocation()
            m[name] = loc
        }
        loc.Add(temp)
    }

    keys := make([]string, 0, len(m))
    for k := range m {
        keys = append(keys, k)
    }
    sort.Strings(keys)

    for _, name := range keys {
        loc := m[name]
        mean := loc.sum / float64(loc.count)
        fmt.Printf("%s: %.1f/%.1f/%.1f\n", name, loc.min, mean, loc.max)
    }
}

Die Ausführung dieser einfacheren Version dauerte etwa 97 Sekunden.

Optimierung der Konvertierung von Strings in Floats

Bei der Analyse des Ausführungsprofils wurde mir klar, dass einer der größten Engpässe die Funktion strconv.ParseFloat war. Im Grunde betrug die Gesamtausführungszeit 23 Sekunden (ungefähr 23 % der Gesamtzeit).

Das Problem bei dieser Funktion besteht darin, dass sie generisch ist, das heißt, dass sie mit jedem gültigen Float funktioniert. Unsere Daten haben jedoch ein sehr spezifisches Temperaturformat. Siehe das Beispiel unten:

Antananarivo;15.6
Iqaluit;-20.7
Dolisie;5.8
Kuopio;-6.8

Das Temperaturformat ist immer dasselbe: eine oder zwei Ziffern vor dem Punkt und eine Ziffer nach dem Punkt und kann am Anfang ein Minuszeichen enthalten. Auf diese Weise können wir eine Funktion erstellen, die Werte auf eine bestimmte Weise umwandelt und so den Prozess optimiert, ohne dass alle generischen Prüfungen von ParseFloat durchgeführt werden müssen.

func bytesToTemp(b []byte) float64 {
    var v int16
    var isNeg int16 = 1

    for i := 0; i < len(b)-1; i++ {
        char := b[i]
        if char == '-' {
            isNeg = -1
        } else if char == '.' {
            digit := int16(b[i+1] - '0')
            v = v*10 + digit
        } else {
            digit := int16(char - '0')
            v = v*10 + digit
        }
    }

    return float64(v*isNeg) / 10
}

Um die Daten im Byte-Format statt im String-Format zu lesen, habe ich die Rückgabe des Scanners von String in Bytes geändert

line := scanner.Bytes()
before, after, _ := bytes.Cut(line, []byte{';'})
name := string(before)
temp := bytesToTemp(after)

Durch diese kleinen Änderungen wurde die Ausführungszeit auf 75 Sekunden gesenkt.

Lesen größerer Datenmengen

Der größte Vorteil der Verwendung von Scan besteht darin, dass das Programm nicht die gesamte Datei auf einmal in den Speicher laden muss. Stattdessen können Sie Zeile für Zeile lesen und Leistung gegen Speicher eintauschen.

Es ist wichtig zu beachten, dass es einen Kompromiss zwischen dem Lesen einer Zeile nach dem anderen und dem gleichzeitigen Laden von 14 GB Daten gibt. Dieser Mittelweg besteht aus dem Lesen von Chunks, also Erinnerungsstücken. Auf diese Weise können wir Blöcke von 128 MB lesen, anstatt die gesamte Datei auf einmal zu lesen.

buf := make([]byte, chunkSize)
reader := bufio.NewReader(file)
var leftData []byte
for {
    n, err := reader.Read(buf)
    if err != nil {
        if err == io.EOF {
            break
        }
        panic(err)
    }

    chunk := append(leftData, buf[:n]...)
    lastIndex := bytes.LastIndex(chunk, []byte{'\n'})
    leftData = chunk[lastIndex+1:]

    lines := bytes.Split(chunk[:lastIndex], []byte{'\n'})

    for _, line := range lines {
        before, after, _ := bytes.Cut(line, []byte{';'})
        name := string(before)
        temp := bytesToTemp(after)

        loc, ok := m[name]
        if !ok {
            loc = NewLocation()
            m[name] = loc
        }
        loc.Add(temp)
    }
}

Dadurch sank die Ausführungszeit auf 70 Sekunden. Besser als zuvor, aber es gibt noch Raum für Verbesserungen.

Datentypen ändern

Fakt ist, dass sich die gesamte Herausforderung um Zahlen mit Nachkommastellen dreht. Allerdings ist der Umgang mit Gleitkommazahlen immer eine große Herausforderung (siehe IEEE-754). Warum also nicht ganze Zahlen zur Darstellung der Temperatur verwenden?

type Location struct {
    min   int16
    max   int16
    sum   int32
    count int32
}

Wie zuvor definiert, wird eine Temperatur immer mit bis zu drei Ziffern dargestellt. Daher können die Werte durch Entfernen des Kommas zwischen -999 und 999 variieren, sodass int16 ausreicht, um sie darzustellen. Zum Summieren und Zählen ist int32 mehr als ausreichend, da dieser Typ zwischen -2147483648 und 2147483647 variieren kann.

Dado que agora esperamos um valor inteiro de 16 bits para a temperatura, precisamos modificar a função bytesToTemp. Para isso, mudamos o retorno para int16 e removemos a divisão no final. Assim, a função vai sempre vai retornar um número inteiro.

func bytesToTemp(b []byte) int16 {
    var v int16
    var isNeg int16 = 1

    for i := 0; i < len(b)-1; i++ {
        char := b[i]
        if char == '-' {
            isNeg = -1
        } else if char == '.' {
            digit := int16(b[i+1] - '0')
            v = v*10 + digit
        } else {
            digit := int16(char - '0')
            v = v*10 + digit
        }
    }

    return v * isNeg
}

Para finalizar, modifiquei a função Add para aceitar os valores inteiros e ajustei o print para dividir os valores antes de mostrá-los na tela. Com isso, o tempo caiu três segundos, indo para 60 segundos. Não é muito, mas uma vitória é uma vitória.

Melhorando a Performance da Conversão de Bytes para String

Novamente analisando o profile, vi que tinha uma certa função chamada slicebytetostring que custava 13,5 segundos de tempo de execução. Analisando, descobri que essa função é a responsável por converter um conjunto de bytes em uma string (o próprio nome da função deixa claro isso). No caso, essa é a função chamada internamente quando se usa a função string(bytes).

Em Go, assim como na maioria das linguagens, strings são imutáveis, o que significa que não podem ser modificadas após serem criadas (normalmente, quando se faz isso, uma nova string é criada). Por outro lado, listas são mutáveis. Ou seja, quando se faz uma conversão de uma lista de bytes para string, é preciso criar uma cópia da lista para garantir que a string não mude se a lista mudar.

Para evitar o custo adicional de alocação de memória nessas conversões, podemos utilizar a biblioteca unsafe para realizar a conversão de bytes para string (Nota: ela é chamada de unsafe por um motivo).

name := unsafe.String(unsafe.SliceData(before), len(before))

Diferente do caso anterior, a função acima reutiliza os bytes passados para gerar a string. O problema disso é que, se a lista original mudar, a string resultante também será afetada. Embora possamos garantir que isso não ocorrerá neste contexto específico, em aplicações maiores e mais complexas, o uso de unsafe pode se tornar bem inseguro.

Com essa mudança, reduzimos o tempo de execução para 51 segundos. Nada mal.

Reimplementando bytes.Cut

Lembra que eu mencionei que as temperaturas sempre tinham formatos específicos? Então, segundo o profile da execução, que separa a linha em duas partes (nome da cidade e temperatura), custa 5.38 segundos para rodar. E refizermos ela na mão?

Para separar os dois valores, precisamos encontrar onde está o ";". Como a gente já sabe, os valores da temperatura podem ter entre três e cinco caracteres. Assim, precisamos verificar se o caractere anterior aos dígitos é um ";". Simples, não?

idx := 0
if line[len(line)-4] == ';' {
    idx = len(line) - 4
} else if line[len(line)-5] == ';' {
    idx = len(line) - 5
} else {
    idx = len(line) - 6
}

before := line[:idx]
after := line[idx+1:]

Com isso, o tempo de execução foi para 46 segundos, cerca de 5 segundos a menos que antes (quem diria, não é?).

Paralelismo para acelerar o processamento

Todo esse tempo, nosso objetivo foi tornar o código o mais rápido possível em um núcleo. Mudando coisas aqui e ali, diminuímos o tempo de 97 segundos para 46 segundos. Claro, ainda daria para melhorar o tempo sem ter que lidar com paralelismo, mas a vida é curta demais para se preocupar com isso, não é?

Para poder rodar o código em vários núcleos, decidi usar a estrutura de canais nativa do Go. Além disso, também criei um grupo de espera que vai indicar quando o processamento dos dados terminaram.

Vale destacar que workers, nesse caso, é uma constante que define quantas goroutines serão criadas para processar os dados. No meu caso, são 12, visto que eu tenho um processador com 6 núcleos e 12 threads.

chunkChan := make(chan []byte, workers)
var wg sync.WaitGroup
wg.Add(workers)

O próximo passo foi criar as goroutines que serão responsável por receber os dados do canal e processá-los. A lógica de processamento dos dados é semelhante ao modelo single thread.

for i := 0; i < workers; i++ {
    go func() {
        for chunk := range chunkChan {
            lines := bytes.Split(chunk, []byte{'\n'})
            for _, line := range lines {
                before, after := parseLine(line)
                name := unsafe.String(unsafe.SliceData(before), len(before))
                temp := bytesToTemp(after)

                loc, ok := m[name]
                if !ok {
                    loc = NewLocation()
                    m[name] = loc
                }
                loc.Add(temp)
            }
        }
        wg.Done()
    }()
}

Por fim, o código responsável por ler os dados do disco e enviá-los ao canal:

for {
    n, err := reader.Read(buf)
    if err != nil {
        if err == io.EOF {
            break
        }
        panic(err)
    }

    chunk := append(leftData, buf[:n]...)
    lastIndex := bytes.LastIndex(chunk, []byte{'\n'})
    leftData = chunk[lastIndex+1:]
    chunkChan <- chunk[:lastIndex]
}
close(chunkChan)
wg.Wait()

Vale ressaltar que os mapas em Go não são thread-safe. Isso significa que acessar ou alterar dados no mesmo mapa de forma concorrente pode levar a problemas de consistência ou erros. Embora não tenha observado problemas durante meus testes, vale a pena tratar esse problema.

Uma das maneiras de resolver esse problema seria criar um mecanismo de trava para o mapa, permitindo que somente uma goroutine consiga utilizá-lo por vez. Isso, claro, poderia tornar a execução um pouco mais lenta.

A segunda alternativa consiste em criar um mapa para cada uma das goroutines, de modo que não vai existir concorrência entre elas. Por fim, os mapas são colocados em um novo canal e os valores do mapa principal calculados a partir deles. Essa solução ainda vai ter um custo, mas vai ser menor que a anterior.

close(chunkChan)

go func() {
    wg.Wait()
    close(mapChan)
}()

keys := make([]string, 0, 825)

m := map[string]*Location{}
for lm := range mapChan {
    for lk, lLoc := range lm {
        loc, ok := m[lk]
        if !ok {
            keys = append(keys, lk)
            m[lk] = lLoc
            continue
        }

        if lLoc.min < loc.min {
            loc.min = lLoc.min
        }
        if lLoc.max > loc.max {
            loc.max = lLoc.max
        }
        loc.sum += lLoc.sum
        loc.count += lLoc.count
    }
}

Além disso, como o processamento passou a ser distribuído entre diferentes núcleos, diminui o tamanho do chunk de 128 MB para 2 MB. Cheguei nesse número testando vários valores, tendo entre 1 MB e 5 MB os melhores resultando. Na média, 2 MB obteve o melhor desempenho.

Enfim, o nosso tempo de processamento caiu de 46 segundos para... 12 segundos.

Otimizando a quebra de linhas no chunk

Todas as vezes que eu analisava o profile, a função bytes.Split chamava a minha atenção. O tempo de execução dela era de 16 segundos (tempo total, considerando todas as goroutines), o que parece justo, visto que ela é responsável por quebrar um chunk em linhas. No entanto, parecia um trabalho dobrado, dado que ela primeiro quebra as linhas para, em seguida, as linhas serem lidas uma a uma. Por que não fazer ambos ao mesmo tempo?

Minha abordagem foi percorrer o chunk e verificar se o byte atual correspondia a um \n. Dessa forma, consegui percorrer todas as linhas ao mesmo tempo em que as quebrava, processando em seguida.

start := 0
start := 0
for end, b := range chunk {
    if b != '\n' {
        continue
    }
    before, after := parseLine(chunk[start:end])
    // ...
    start = end + 1
}

Com essa simples mudança, o tempo de execução caiu para aproximadamente 9 segundos.

Executed in    8.45 secs    fish           external
   usr time   58.47 secs  152.00 micros   58.47 secs
   sys time    4.52 secs  136.00 micros    4.52 secs

Próximos passos

Atualmente, o maior gargalo da aplicação é o mapa. Somando todas as operações de leitura e escrita, são 32 segundos (de longe, o tempo mais alto). Talvez criar um algoritmo de hash mais eficiente resolva? Fica como ideia para o futuro.

No mais, conseguimos diminuir o tempo de 1 minuto e 40 segundos para quase 8 segundos, sem usar qualquer biblioteca externa. Além disso, tentando fazer a aplicação ficar cada vez mais rápida, me fez aprender muita coisa.

Das obige ist der detaillierte Inhalt vonLösung der Milliarden-Zeilen-Herausforderung in Go (von bis s). Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!