Челлендж по обработке миллиарда строк на Go: от 1 минуты 45 секунд до 4 секунд

Пару недель назад я прочитал о запавшем мне в душу челлендже по обработке миллиарда строк, поэтому захотел решить его на Go.

Я немного опоздал, соревнования проводились в январе. И на Java. Меня не особо интересует Java, зато давно интересует оптимизация кода на Go.

Этот челлендж был очень прост: обработать текстовый файл названий метеорологических станций и температур, и для каждой станции вывести минимальное, среднее и максимальное значение. Чтобы упростить задачу, было ещё несколько ограничений, однако я проигнорировал те, что относятся только к Java.

Вот несколько строк с примером входных данных:

Hamburg;12.0
Bulawayo;8.9
Palembang;38.8
St. John's;15.2
Cracow;12.6
...

Единственная тонкость заключается в том, что входной файл состоит из миллиарда строк. Это примерно 13 ГБ данных. Я уже разобрался, что дисковый ввод-вывод не становится узким местом, обычно замедляют подобные программы распределения памяти и парсинг.

В этой статье описывается девять написанных мной на Go решений, каждое из которых быстрее предыдущего. Первое, простое и идиоматичное, выполняется на моей машине 1 минуту 45 секунд, а последнее — примерно 4 секунды. По ходу дела я буду показывать, как использовал профилировщик Go, чтобы понимать, на что тратится время.

Вот список решений, от самого медленного до самого быстрого:

• r1: простое и идиоматичное

• r2: map со значениями указателей

• r3: ручной парсинг температур

• r4: целые числа с фиксированной запятой

• r5: избегаем bytes.Cut

• r6: избегаем bufio.Scanner

• r7: специальная хэш-таблица

• r8: распараллеленное r1

• r9: распараллеленное r7

Я хотел, чтобы каждое из решений было портируемым кодом на Go, использующим только стандартную библиотеку: никакого ассемблера, никаких unsafe и никаких файлов с отображением в память. Для меня 4 секунды, или 3,2 ГБ/с показались достаточно быстрым результатом. Для сравнения: самое быстрое, сильно оптимизированное решение на Java выполняется на моей машине меньше чем за секунду — неплохо!

Уже существует множество готовых решений на Go и по крайней мере одна хорошая статья. Моё решение быстрее некоторых других, но чуть медленнее, чем самое быстрое. Однако перед написанием своего я не изучал другие — хотел, чтобы мои решения были независимыми.

Если вам интересны только показатели, то перейдите в конец статьи, там есть таблица с результатами.

Отправная точка

Вот несколько исходных показателей для понимания, на что ориентироваться. Во-первых сколько времени нужно для простого считывания 13 ГБ данных при помощи cat:

$ time cat measurements.txt >/dev/null
0m1.052s

Стоит отметить, что это лучший показатель из пяти, то есть я позволил файлу кэшироваться. Кто знает, допускает ли Linux хранить все 13 ГБ в дисковом кэше; предположительно, да, потому что в первый раз это заняло почти 6 секунд.

Для сравнения, выполнение каких-то операций происходит существенно медленнее: wc занимает почти минуту:

$ time wc measurements.txt 
 1000000000  1179173106 13795293380 measurements.txt
0m55.710s

Для создания простого решения этой задачи я, вероятно, начну с AWK. В этом решении используется Gawk, потому что сортировать вывод проще функцией asorti. Я воспользовался опцией -b, чтобы применить режим «символы как байты», что немного ускоряет работу:

$ time gawk -b -f 1brc.awk measurements.txt >measurements.out
7m35.567s

Уверен, что смогу побить показатель в 7 минут даже с простым решением на Go, так что давайте начнём с этого.

Я начну с оптимизации последовательной одноядерной версии (решения 1-7), а затем распараллелю его (решения 8 и 9). Все результаты получены с Go 1.21.5 на ноутбуке с linux/amd64, быстрым SSD-накопителем и 32 ГБ ОЗУ.

Многие из моих решений и большинство самых быстрых решений предполагают, что входные данные всегда валидны. Например, что температуры имеют ровно одно значение после запятой. Многие мои решения приведут к runtime panic или к неверным результатам в случае невалидных входных данных.

Решение 1: простой и идиоматичный Go

Я хотел, чтобы первая версия была простым, незатейливым решением с использованием инструментов только из стандартной библиотеки Go: bufio.Scanner для считывания строк, strings.Cut для разбиения по ';', strconv.ParseFloat для парсинга температур и обычной map Go для накапливания результатов.

Сначала я покажу решение целиком, а потом объясню самые интересные части:

func r1(inputPath string, output io.Writer) error {
    type stats struct {
        min, max, sum float64
        count         int64
    }

    f, err := os.Open(inputPath)
    if err != nil {
        return err
    }
    defer f.Close()

    stationStats := make(map[string]stats)

    scanner := bufio.NewScanner(f)
    for scanner.Scan() {
        line := scanner.Text()
        station, tempStr, hasSemi := strings.Cut(line, ";")
        if !hasSemi {
            continue
        }

        temp, err := strconv.ParseFloat(tempStr, 64)
        if err != nil {
            return err
        }

        s, ok := stationStats[station]
        if !ok {
            s.min = temp
            s.max = temp
            s.sum = temp
            s.count = 1
        } else {
            s.min = min(s.min, temp)
            s.max = max(s.max, temp)
            s.sum += temp
            s.count++
        }
        stationStats[station] = s
    }

    stations := make([]string, 0, len(stationStats))
    for station := range stationStats {
        stations = append(stations, station)
    }
    sort.Strings(stations)

    fmt.Fprint(output, "{")
    for i, station := range stations {
        if i > 0 {
            fmt.Fprint(output, ", ")
        }
        s := stationStats[station]
        mean := s.sum / float64(s.count)
        fmt.Fprintf(output, "%s=%.1f/%.1f/%.1f", station, s.min, mean, s.max)
    }
    fmt.Fprint(output, "}n")
    return nil
}

Это базовое решение обрабатывает один миллиард строк за 1 минуту 45 секунд. Точно лучше, чем 7 минут решения на AWK.

Решение 2: map со значениями указателей

Я обучался на создании своей программы count-words, в которой происходит чуть больше хэширования, чем нужно. В каждой строке мы хэшируем строку символов дважды: сначала когда пытаемся получить значение из map, а потом когда обновляем map.

Чтобы избежать этого, можно использовать map[string]stats (значения указателей) и обновлять адресуемую указателем struct, вместо map[string]stats и обновления самой хэш-таблицы.

Однако сначала я хотел убедиться в этом при помощи профилировщика Go. Чтобы добавить профилирование CPU к программе на Go, достаточно всего нескольких строк.

$ ./go-1brc -cpuprofile=cpu.prof -revision=1 measurements-10000000.txt >measurements-10000000.out
Processed 131.6MB in 965.888929ms
$ go tool pprof -http=: cpu.prof
...

Эти команды создали следующий профиль решения 1, пропущенного через урезанный входной файл в 10 миллионов строк:

Операции с map занимают целых 30% времени: 12,24% на присвоение и 17,35% на поиск. Использовав значение указателя, мы должны избавиться от основной части времени на присвоения map.

Примечание: это изображение профиля также демонстрирует, на что тратится остальная часть времени:

• Сканирование строк при помощи Scanner.Scan

• Поиск ';' при помощи strings.Cut

• Парсинг температуры при помощи strconv.ParseFloat

• Вызов Scanner.Text, распределяющего строку символов для строки файла

Как бы то ни было, моё второе решение стало лишь небольшим улучшением операций с map:

stationStats := make(map[string]*stats)
scanner := bufio.NewScanner(f)
for scanner.Scan() {
    // ...
    s := stationStats[station]
    if s == nil {
        stationStats[station] = &stats{
            min:   temp,
            max:   temp,
            sum:   temp,
            count: 1,
        }
    } else {
        s.min = min(s.min, temp)
        s.max = max(s.max, temp)
        s.sum += temp
        s.count++
    }
}

В общем случае, когда станция существует в map, мы теперь выполняем только одну операцию с map, s := stationStats[station], чтобы хэширование названия станции и доступ к хэш-таблице должно было выполняться только один раз. Если оно уже есть в map (общий случай для одного миллиарда строк), то мы обновляем существующую struct, адресуемую указателем.

Это не особо помогает, но что-то даёт: использование значений указателей в map снижает время выполнения с 1 минуты 45 секунд до 1 минуты 31 секунды.

Решение 3: избегаем strconv.ParseFloat

В третьем решении всё наконец становится более хардкорным: мы будем парсить температуру при помощи собственного кода, а не через strconv.ParseFloat. Функция стандартной библиотеки обрабатывает кучу пограничных случаев, которые для простых температур входных данных нам поддерживать не нужно: у нас будет всего две или три цифры в формате 1.2 или  34.5 (и некоторые с минусом перед числом).

Кроме того, strconv.ParseFloat получает аргумент string, и теперь, когда мы её не используем, можно обойтись байтовым срезом непосредственно из Scanner.Bytes, а не распределять и копировать строку при помощи Scanner.Text.

Теперь мы парсим температуру вот так:

negative := false
index := 0
if tempBytes[index] == '-' {
    index++
    negative = true
}
temp := float64(tempBytes[index] - '0') // парсим первую цифру
index++
if tempBytes[index] != '.' {
    temp = temp*10 + float64(tempBytes[index]-'0') // парсим опциональную вторую цифру
    index++
}
index++ // skip '.'
temp += float64(tempBytes[index]-'0') / 10 // парсим десятичную цифру
if negative {
    temp = -temp
}

Не особо красиво, но и сложного ничего нет. Так мы снизили время с 1 минуты 31 секунды до менее чем минуты: 55,8 секунды.

Решение 4: целые числа с фиксированной запятой

В былые времена команды с плавающей запятой были гораздо медленнее, чем целочисленные. Сегодня они лишь немного медленнее, но, вероятно, стоит по возможности избегать их.

В нашей задаче каждая температура имеет один разряд после запятой, так что для её описания можно легко использовать целые числа с фиксированной запятой. Например, мы можем представить 34,5 как целочисленное 345. И только в самом конце, перед непосредственным выводом результатов, мы преобразуем их обратно во float.

То есть моё четвёртое решение, по сути, такое же, как и решение 3, но со следующим полем struct stats:

type stats struct {
    min, max, count int32
    sum             int64
}

Перед выводом результатов его нужно разделить на 10:

mean := float64(s.sum) / float64(s.count) / 10
fmt.Fprintf(output, "%s=%.1f/%.1f/%.1f",
    station, float64(s.min)/10, mean, float64(s.max)/10)

Для минимальных и максимальных температур я использовал 32-битные integer, так как максимум, вероятно, будет примерно 500 (50 градусов Цельсия). Можно использовать int16, но из прежнего опыта я сделал вывод, что современные 64-битные CPU чуть медленнее работают с 16-битными integer, чем с 32-битными. В моих тестах они не показали какой-то значимой разницы, но я всё равно выбрал 32-битные.

Использование integer снизило время с 55,8 секунды до 51,0 секунды — небольшой выигрыш.

Решение 5: избегаем bytes.Cut

Чтобы создать решение 5, я записал ещё один профиль (решения 4):

Итак, всё стало сложнее. Операции map доминируют, и переход к специализированной хэш-таблице может оказаться немного неочевидным. Поэтому мы избавимся от bufio.Scanner. Давайте попрокрастинируем и избавимся от bytes.Cut.

Я подумал, что это простой способ экономии времени. Взгляните на этот пример строки:

New Orleans;11.7

Будет быстрее парсить температуру с конца и находить ';' там, чем сканировать всё название станции в поисках ';'. Этот довольно уродливый код делает именно это:

end := len(line)
tenths := int32(line[end-1] - '0')
ones := int32(line[end-3] - '0') // line[end-2] is '.'
var temp int32
var semicolon int
if line[end-4] == ';' {          // положительная температура N.N
    temp = ones*10 + tenths
    semicolon = end - 4
} else if line[end-4] == '-' {   // отрицательная температура -N.N
    temp = -(ones*10 + tenths)
    semicolon = end - 5
} else {
    tens := int32(line[end-4] - '0')
    if line[end-5] == ';' {      // положительная температура NN.N
        temp = tens*100 + ones*10 + tenths
        semicolon = end - 5
    } else {                     // отрицательная температура -NN.N
        temp = -(tens*100 + ones*10 + tenths)
        semicolon = end - 6
    }
}
station := line[:semicolon]

Отказавшись от bytes.Cut, мы снизили время с 51,0 секунды до 46,0 секунды — ещё одна маленькая победа.

Решение 6: избегаем bufio.Scanner

Теперь мы попробуем избавиться от bufio.Scanner. Задумайтесь: чтобы найти конец каждой строки файла, сканеру приходится обходить все байты в поисках символа переноса строки. Затем мы снова обрабатываем множество байтов, чтобы спарсить температуру и найти ';'. Так что давайте попробуем объединить эти этапы и избавимся от bufio.Scanner.

В решении 6 мы распределяем буфер на 1 МБ для чтения файла большими блоками, ищем последний символ переноса строки в блоке, чтобы убедиться, что мы не разрезаем строку пополам, а затем обрабатываем каждый блок. Это выглядит так:

buf := make([]byte, 1024*1024)
readStart := 0
for {
    n, err := f.Read(buf[readStart:])
    if err != nil && err != io.EOF {
        return err
    }
    if readStart+n == 0 {
        break
    }
    chunk := buf[:readStart+n]

    newline := bytes.LastIndexByte(chunk, 'n')
    if newline < 0 {
        break
    }
    remaining := chunk[newline+1:]
    chunk = chunk[:newline+1]

    for {
        station, after, hasSemi := bytes.Cut(chunk, []byte(";"))
        // ... дальше та же обработка температуры, что и в r4 ...

Устранениеbufio.Scanner и выполнение собственного сканирования снизило время с 46,0 секунды до 41,3 секунды. Ещё одна крошечная победа, но мы примем её.

Решение 7: специальная хэш-таблица

На решении 7 шутки заканчиваются. Мы реализуем собственную хэш-таблицу вместо map Go. У этого подхода есть два преимущества:

1. Мы сможем хэшировать название станции в процессе поиска ';', избегая повторной обработки байтов.

2. Мы можем хранить каждый ключ в нашей хэш-таблице как байтовый срез, избегая необходимости преобразовывать каждый ключ в string (при этом выполняется распределение и копирование для каждой строки файла).

Я писал о том, как реализовать хэш-таблицу на C, но также я реализовал и собственную «счётную» хэш-таблицу на Go, откуда и взял эту реализацию.

Это простая реализация, использующая алгоритм хэширования FNV-1a с линейным зондированием: если возникает коллизия, задействуется следующий пустой слот.

Чтобы упростить, я заранее распределяю большое количество хэш-бакетов (я использовал 100000), чтобы избежать необходимости написания логики изменения размера страницы. Если таблица заполнится больше, чем наполовину, у кода произойдёт panic. Я замерил, что мы получим примерно 2% коллизий хэшей.

На этот раз кода гораздо больше — подготовка хэш‑таблицы, само хэширование, зондирование таблицы и вставка:

// Структура хэш-таблицы:
type item struct {
    key  []byte
    stat *stats
}
items := make([]item, 100000) // хэш-бакеты с линейным зондированием
size := 0                     // количество активных элементов в срезе элементов

buf := make([]byte, 1024*1024)
readStart := 0
for {
    // ... то же разбиение на блоки, что и в r6 ...

    for {
        const (
            // 64-битные константы FNV-1 из hash/fnv.
            offset64 = 14695981039346656037
            prime64  = 1099511628211
        )

        // Хэшируем название станции и ищем ';'.
        var station, after []byte
        hash := uint64(offset64)
        i := 0
        for ; i < len(chunk); i++ {
            c := chunk[i]
            if c == ';' {
                station = chunk[:i]
                after = chunk[i+1:]
                break
            }
            hash ^= uint64(c) // FNV-1a is XOR then *
            hash *= prime64
        }
        if i == len(chunk) {
            break
        }

        // ... тот же парсинг температур, что и в r6 ...

        // Переходим к нужному бакету в хэш-таблице.
        hashIndex := int(hash & uint64(len(items)-1))
        for {
            if items[hashIndex].key == nil {
                // Найден пустой слот, добавляем новый элемент (копируем ключ).
                key := make([]byte, len(station))
                copy(key, station)
                items[hashIndex] = item{
                    key: key,
                    stat: &stats{
                        min:   temp,
                        max:   temp,
                        sum:   int64(temp),
                        count: 1,
                    },
                }
                size++
                if size > len(items)/2 {
                    panic("too many items in hash table")
                }
                break
            }
            if bytes.Equal(items[hashIndex].key, station) {
                // Найден совпадающий слот, прибавляем к имеющейся статистике.
                s := items[hashIndex].stat
                s.min = min(s.min, temp)
                s.max = max(s.max, temp)
                s.sum += int64(temp)
                s.count++
                break
            }
            // Слот уже содержит другой ключ, пробуем следующий слот (линейное зондирование).
            hashIndex++
            if hashIndex >= len(items) {
                hashIndex = 0
            }
        }
    }

    readStart = copy(buf, remaining)
}

Отдача от всего этого кода велика: специальная хэш-таблица снижает время с 41,3 до 25,8 секунды.

Решение 8: параллельная обработка блоков

В решении 8 я хотел добавить параллелизм. Однако я решил вернуться к простому и идиоматичному коду из первого решения, использующему bufio.Scanner и strconv.ParseFloat, распараллелив его. Так мы увидим, что даёт лучшие результаты, оптимизация или параллелизация, а в девятом решении реализуем и то, и другое.

Задачу map-reduce параллелизировать очень легко: разбить файл на блоки схожего размера (по одному на каждое ядро CPU), запустить поток (в Go горутину) для обработки каждого блока, а в конце объединить результаты.

Вот, как это выглядит на высоком уровне:

// Определяем непересекающиеся части для разбиения файла (каждая часть имеет смещение и размер).
parts, err := splitFile(inputPath, maxGoroutines)
if err != nil {
    return err
}

// Запускаем горутину для обработки каждой части, возвращая результаты в канал.
resultsCh := make(chan map[string]r8Stats)
for _, part := range parts {
    go r8ProcessPart(inputPath, part.offset, part.size, resultsCh)
}

// Ждём возврата результатов и агрегируем их.
totals := make(map[string]r8Stats)
for i := 0; i < len(parts); i++ {
    result := <-resultsCh
    for station, s := range result {
        ts, ok := totals[station]
        if !ok {
            totals[station] = r8Stats{
                min:   s.min,
                max:   s.max,
                sum:   s.sum,
                count: s.count,
            }
            continue
        }
        ts.min = min(ts.min, s.min)
        ts.max = max(ts.max, s.max)
        ts.sum += s.sum
        ts.count += s.count
        totals[station] = ts
    }
}

Функция splitFile довольно скучна, поэтому я не стал её сюда включать. Она смотрит на размер файла, делит его на нужное нам количество частей, а затем ищет каждую часть, считывая 100 байтов перед концом и находя последний символ переноса строки, чтобы гарантировать, что каждая часть заканчивается полной строкой файла.

Функция r8ProcessPart, по сути, такая же, как и в решении r1, но она начинает с перехода к смещению части и ограничивает длину размером части (при помощи io.LimitedReader). Закончив, она отправляет собственную map статистики обратно в канал:

func r8ProcessPart(inputPath string, fileOffset, fileSize int64,
                   resultsCh chan map[string]r8Stats) {
    file, err := os.Open(inputPath)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    _, err = file.Seek(fileOffset, io.SeekStart)
    if err != nil {
        panic(err)
    }
    f := io.LimitedReader{R: file, N: fileSize}

    stationStats := make(map[string]r8Stats)

    scanner := bufio.NewScanner(&f)
    for scanner.Scan() {
        // ... та же обработка, что и в r1 ...
    }

    resultsCh <- stationStats
}

Параллельная обработка входного файла обеспечивает существенный выигрыш по сравнению с r1, снижая время с 1 минуты 45 секунд до 24,3 секунды. Для сравнения: предыдущая «оптимизированная непараллельная» версия (решение 7) занимала 25,8 секунды. То есть в этом случае параллелизация чуть быстрее оптимизации, к тому же намного проще.

Решение 9: все оптимизации плюс параллелизация

В решении 9, нашей последней попытке, мы просто соединим все оптимизации с r1 по r7 с параллелизацией, реализованной в r8.

Я использовал ту же функцию splitFile из r8, а остальную часть кода просто скопировал из r7, поэтому здесь нет ничего нового. За исключением результатов: эта окончательная версия снизила время с 24,3 секунды до 3,99 секунды — огромная победа.

Любопытно, что поскольку вся реальная обработка теперь находится в одной большой функции r9ProcessPart, граф профиля больше не особо полезен. Вот как он теперь выглядит:

Как видите, 82% времени тратится на r9ProcessPart, bytes.Equal занимает 13%, а считывание файла занимает оставшиеся 5%.

Если мы хотим добиться более подробного профилирования, нам нужно уйти глубже, чем уровень функций, который нам показывает режим графа, и использовать режим исходников. Вот внутренний цикл:

Меня этот отчёт сбил с толку. Почему в нём показано, что if items[hashIndex].key == nil занимает 5,01 с, но у вызова bytes.Equal показано всего 390 мс? Поиск среза ведь гораздо менее затратен, чем вызов функции? Если вы специалист по производительности Go и поможете мне это проинтерпретировать, то напишите!

Как бы то ни было, я уверен, что можно придумать и более безумные оптимизации, но я решил остановиться на этом. Для меня обработка миллиарда строк за 4 секунды, или по 250 миллиона строк в секунду — это вполне достаточно.

Таблица результатов

Ниже приведена таблица со всеми моими решениями на Go, а также с самыми быстрыми решениями на Go и на Java. Каждый результат — лучший из пяти прогонов решения с одинаковым входным файлом из миллиарда строк.

Я приблизительно в том же интервале, что и версия на Go Александра Ястребова (AY). Его решение похоже на моё: разбиваем файл на блоки, используем специальную хэш-таблицу (он даже как и я использовал хэширование FNV) и парсим температуру как integer. Однако он использует файлы с отображением в память, от которых я отказался из соображений портируемости. Вероятно, поэтому его версия немного быстрее.

Томас Вюртенгер (при помощи других разработчиков) создал в исходном челлендже на Java самое быстрое решение. На моей машине оно выполняется меньше, чем за секунду, в четыре раза быстрее, чем моя версия на Go. Похоже, что наряду с параллельной обработкой и отображением файлов в память он использовал развёрнутые циклы, код парсинга без ветвления и другие низкоуровневые трюки.

Кажется, Томас — основатель и важный контрибьютор в GraalVM (быстрой Java Virtual Machine с компиляцией перед исполнением). Так что он определённо специалист в этой области. Отличная работа, Томас!

Последние комментарии

Почему всё это важно?

В случае большинства повседневных программных задач лучше всего начинать с простого и идиоматичного кода. Если вы вычисляете статистику для миллиарда температур, а ответ нужен лишь один раз, то 1 минуты 45 секунд, вам, вероятно, будет достаточно.

Но если вы создаёте конвейер обработки данных и сможете ускорить код в четыре, а то и в 26 раз, то вы не только порадуете пользователей, но и сможете серьёзно сэкономить в затратах на вычислительные ресурсы — если система хорошо загружена, то затраты на вычисления могут быть в 4 или 26 раза меньше исходных!

А если вы создаёте среду исполнения наподобие GraalVM или интерпретатор наподобие GoAWK, то такой уровень производительности очень важен: если вы ускорите интерпретатор, то и программы пользователей тоже будут выполняться гораздо быстрее.

Кроме того, просто интересно писать код, выжимающий из машины максимум.