The fundamental rule for large data is: never load more than you need into memory at once. Loading a 10 GB file with os.ReadFile allocates 10 GB of heap, triggers the GC constantly, and likely crashes your process. Streaming the same file processes it in kilobytes of working memory regardless of file size.
This guide covers the patterns that keep memory usage flat as data volume grows — streaming reads, chunked parallel processing, buffer reuse with sync.Pool, and database cursor patterns.
For file I/O fundamentals see Go file system operations. For worker pool patterns see Go worker pools.
Streaming vs Loading: The Core Distinction
os.ReadFile reads the entire file into one byte slice. For anything larger than available RAM, this is wrong:
// ❌ Loads entire file into memory — O(file size) memory
data, _ := os.ReadFile("events.log")
lines := strings.Split(string(data), "\n")
for _, line := range lines { process(line) }
// ✅ Constant memory regardless of file size
f, _ := os.Open("events.log")
defer f.Close()
scanner := bufio.NewScanner(f)
for scanner.Scan() {
process(scanner.Text())
}
bufio.Scanner maintains a small internal buffer (default 64 KB). Lines are processed and discarded as they’re read. A 100 GB log file uses the same ~64 KB of scanner buffer as a 1 KB file.
When lines can exceed 64 KB (common in JSON-per-line or base64 data), increase the buffer:
scanner := bufio.NewScanner(f)
scanner.Buffer(make([]byte, 0, 1<<20), 64<<20) // up to 64 MB per line
Always check scanner.Err() after the loop — it’s nil on clean EOF but non-nil if a read error occurred mid-stream:
for scanner.Scan() {
process(scanner.Text())
}
if err := scanner.Err(); err != nil {
return fmt.Errorf("scanning: %w", err)
}
Parallel Processing with a Worker Pool
Sequential streaming processes one item at a time. When processing is CPU-bound (parsing, transformation, hashing), parallelizing across multiple goroutines multiplies throughput proportionally to CPU count.
The pattern: a producer goroutine reads and sends items; a pool of worker goroutines consumes and processes them:
func processFileParallel(ctx context.Context, path string, numWorkers int, fn func(string) error) error {
f, err := os.Open(path)
if err != nil {
return err
}
defer f.Close()
lines := make(chan string, numWorkers*2) // buffer prevents producer/worker stalls
// Producer: read lines and send to channel
var scanErr error
go func() {
defer close(lines)
scanner := bufio.NewScanner(f)
for scanner.Scan() {
select {
case lines <- scanner.Text():
case <-ctx.Done():
return
}
}
scanErr = scanner.Err()
}()
// Workers: process lines concurrently
var wg sync.WaitGroup
errc := make(chan error, numWorkers)
for i := 0; i < numWorkers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for line := range lines {
if err := fn(line); err != nil {
errc <- err
return
}
}
}()
}
wg.Wait()
close(errc)
if scanErr != nil {
return scanErr
}
for err := range errc {
return err // return first worker error
}
return nil
}
For CPU-bound processing, numWorkers = runtime.NumCPU() is the right starting point. For I/O-bound processing (each item triggers a network call or database query), use more workers — try 2×–5× CPU count and measure.
Chunked Processing for Bulk Operations
Some operations are more efficient in batches — database inserts, API calls, transformations that benefit from locality. Batching reduces per-item overhead:
func processInChunks[T any](ctx context.Context, source <-chan T, chunkSize int, fn func([]T) error) error {
chunk := make([]T, 0, chunkSize)
flush := func() error {
if len(chunk) == 0 {
return nil
}
err := fn(chunk)
chunk = chunk[:0] // reset slice, keep capacity
return err
}
for {
select {
case item, ok := <-source:
if !ok {
return flush() // process remaining items on channel close
}
chunk = append(chunk, item)
if len(chunk) >= chunkSize {
if err := flush(); err != nil {
return err
}
}
case <-ctx.Done():
return ctx.Err()
}
}
}
Add a time-based flush (like a 100ms ticker) if you want to ensure partial chunks are processed promptly rather than waiting for the chunk to fill. This is the same pattern used in log aggregators and metrics collectors.
Reusing Buffers with sync.Pool
If each item in your processing pipeline requires a temporary buffer (for parsing, transformation, or encoding), allocating a new buffer per item puts significant pressure on the GC. sync.Pool amortizes this by reusing buffers across items:
var bufPool = sync.Pool{
New: func() any {
return bytes.NewBuffer(make([]byte, 0, 4096))
},
}
func processRecord(data []byte) ([]byte, error) {
buf := bufPool.Get().(*bytes.Buffer)
buf.Reset() // clear contents, keep capacity
defer bufPool.Put(buf)
// Use buf for intermediate work
if err := json.NewEncoder(buf).Encode(transform(data)); err != nil {
return nil, err
}
// Copy the result — buf goes back to the pool, result lives independently
result := make([]byte, buf.Len())
copy(result, buf.Bytes())
return result, nil
}
Always reset the buffer before use (buf.Reset()), not before returning — in case the caller returns early from an error. Always defer bufPool.Put(buf) immediately after Get to ensure the buffer is returned even on errors.
The GC can collect objects from the pool at any time (typically at GC cycles). Don’t store anything in a pool object that needs to outlive the pool’s borrow period.
Database Cursor Pattern
Querying millions of rows with db.Query + iterating rows.Next() is already streaming — the database driver fetches rows in batches internally. But for very large result sets, you can page explicitly to control memory and allow resumability:
func processAllUsers(ctx context.Context, db *sql.DB, fn func(*User) error) error {
const pageSize = 1000
var lastID int64 = 0
for {
rows, err := db.QueryContext(ctx, `
SELECT id, name, email FROM users
WHERE id > $1
ORDER BY id ASC
LIMIT $2
`, lastID, pageSize)
if err != nil {
return fmt.Errorf("query: %w", err)
}
count := 0
for rows.Next() {
var u User
if err := rows.Scan(&u.ID, &u.Name, &u.Email); err != nil {
rows.Close()
return err
}
if err := fn(&u); err != nil {
rows.Close()
return err
}
lastID = u.ID
count++
}
rows.Close()
if err := rows.Err(); err != nil {
return err
}
if count < pageSize {
break // last page
}
}
return nil
}
This cursor-based pagination (keyset pagination) is more efficient than LIMIT/OFFSET for large tables — OFFSET N forces the database to skip N rows, while WHERE id > lastID uses an index directly.
Measuring Memory Usage
Profile memory before and after changes to confirm improvements:
import (
"runtime"
"fmt"
)
func printMemStats(label string) {
var m runtime.MemStats
runtime.ReadMemStats(&m)
fmt.Printf("[%s] alloc=%.1f MB, total_alloc=%.1f MB, sys=%.1f MB, gc=%d\n",
label,
float64(m.Alloc)/1e6,
float64(m.TotalAlloc)/1e6,
float64(m.Sys)/1e6,
m.NumGC,
)
}
printMemStats("before")
processLargeFile(...)
printMemStats("after")
Alloc is current heap in use. TotalAlloc is cumulative — if it grows much faster than Alloc, you’re allocating and discarding many objects (high GC pressure). NumGC increasing rapidly signals the same.
For a full heap profile, use pprof:
go tool pprof -alloc_space http://localhost:6060/debug/pprof/heap
The heap profile shows which functions allocated the most memory — the starting point for any memory optimization.
Summary
- Stream files with
bufio.Scanner— constant memory regardless of file size; always checkscanner.Err()after the loop - Parallelize CPU-bound work with a worker pool sized at
runtime.NumCPU(); use more workers for I/O-bound work - Batch items into chunks when per-item overhead is high (DB inserts, API calls)
- Reuse buffers with
sync.Poolto reduce GC pressure in high-throughput processing - Use keyset pagination (
WHERE id > lastID) for database cursor patterns — faster than OFFSET for large tables - Profile with
runtime.ReadMemStatsandpprofbefore optimizing — measure, then act
Comments