Large Data Handling

Learn strategies for processing large datasets efficiently with memory and performance optimizations.

Generate Test Data

First, let's create a challenging CSV file for testing:

package main
import (
    "encoding/csv"
    "fmt"
    "math/rand"
    "os"
    "strconv"
    "time"
)

func generateLargeCSV(filename string, records int) error {
    file, err := os.Create(filename)
    if err != nil {
        return err
    }
    defer file.Close()

    writer := csv.NewWriter(file)
    defer writer.Flush()

    // Write header
    header := []string{"ID", "Name", "Email", "Age", "Salary",
                      "Department", "City", "Country", "JoinDate", "Active"}
    writer.Write(header)

    departments := []string{"Engineering", "Sales", "Marketing",
                           "HR", "Finance", "Operations"}
    cities := []string{"New York", "London", "Tokyo", "Paris",
                       "Berlin", "Sydney", "Toronto", "Singapore"}

    rand.Seed(time.Now().UnixNano())

    start := time.Now()
    for i := 1; i <= records; i++ {
        record := []string{
            strconv.Itoa(i),
            fmt.Sprintf("Employee_%d", i),
            fmt.Sprintf("emp%d@company.com", i),
            strconv.Itoa(22 + rand.Intn(43)),  // Age 22-65
            fmt.Sprintf("%.2f", 30000.0 + rand.Float64()*120000.0),
            departments[rand.Intn(len(departments))],
            cities[rand.Intn(len(cities))],
            "USA",
            fmt.Sprintf("2020-01-%02d", 1+rand.Intn(28)),
            strconv.FormatBool(rand.Float64() > 0.2),  // 80% active
        }
        writer.Write(record)

        if i % 10000 == 0 {
            fmt.Printf("Generated %d records...\n", i)
        }
    }

    duration := time.Since(start)
    fileInfo, _ := file.Stat()
    fmt.Printf("\nGenerated %d records in %v\n", records, duration)
    fmt.Printf("File size: %.2f MB\n", float64(fileInfo.Size())/(1024*1024))

    return nil
}

generateLargeCSV("employees.csv", 100000)

Result

Generated 10000 records...
Generated 20000 records...
...
Generated 100000 records...

Generated 100000 records in 154.4ms
File size: 8.87 MB

Strategy 1: Load All Data (Memory Heavy)

The simplest approach - load everything into memory:

type Employee struct {
    ID         int
    Department string
    Salary     float64
    Active     bool
}

func printMemStats(label string) {
    var m runtime.MemStats
    runtime.ReadMemStats(&m)
    fmt.Printf("%s: %.2f MB\n", label, float64(m.Alloc)/(1024*1024))
}

func loadAll(filename string) ([]Employee, error) {
    file, _ := os.Open(filename)
    defer file.Close()

    reader := csv.NewReader(file)
    records, _ := reader.ReadAll()  // Load all at once

    employees := make([]Employee, 0, len(records)-1)
    for i, rec := range records {
        if i == 0 { continue }  // Skip header

        id, _ := strconv.Atoi(rec[0])
        salary, _ := strconv.ParseFloat(rec[4], 64)
        employees = append(employees, Employee{
            ID: id, Department: rec[5],
            Salary: salary, Active: rec[9] == "true",
        })
    }

    return employees, nil
}

runtime.GC()
printMemStats("Start")

employees, _ := loadAll("test_data.csv")
printMemStats("After Load")

// Filter with plyGO
start := time.Now()
result := plygo.From(employees).
    Where("Department").Equals("Engineering").
    Where("Salary").GreaterThan(80000.0).
    Collect()

fmt.Printf("Filter time: %v\n", time.Since(start))
fmt.Printf("Found: %d of %d\n", len(result), len(employees))
printMemStats("After Filter")

Result

Start: 0.11 MB
After Load: 2.32 MB
Filter time: 3.37s
Found: 2015 of 10000
After Filter: 3.08 MB

Analysis:

• ⚠️ High memory usage: 2.32 MB for 10k records
• ⚠️ Slow filtering: 3.37s (plyGO processes all data)
• ✅ Simple code: Easy to understand
• ❌ Not scalable: Won't work for millions of records

Strategy 2: Streaming with Indices (Memory Efficient)

Stream data and store only matching indices:

func streamWithIndices(filename string) ([]int, error) {
    file, _ := os.Open(filename)
    defer file.Close()

    reader := csv.NewReader(file)
    reader.Read()  // Skip header

    var matchingIndices []int
    rowNum := 0

    for {
        record, err := reader.Read()
        if err == io.EOF {
            break
        }
        rowNum++

        // Filter inline while reading
        department := record[1]
        salary, _ := strconv.ParseFloat(record[2], 64)

        if department == "Engineering" && salary > 80000.0 {
            matchingIndices = append(matchingIndices, rowNum)
        }
    }

    return matchingIndices, nil
}

runtime.GC()
printMemStats("Start")

start := time.Now()
indices, _ := streamWithIndices("test_data.csv")
fmt.Printf("Stream time: %v\n", time.Since(start))
fmt.Printf("Found: %d indices\n", len(indices))

printMemStats("After Stream")

Result

Start: 0.11 MB
Stream time: 3.52ms
Found: 2015 indices
After Stream: 1.06 MB

Analysis:

• ✅ Low memory: Only 1.06 MB (54% less than load-all)
• ✅ Fast: 3.52ms (957x faster!)
• ✅ Scalable: Can handle very large files
• ⚠️ Indices only: Need second pass to get full records
• ✅ Best for: Finding what to process, then loading selectively

Strategy 3: Chunked Processing (Balanced)

Process data in chunks to balance memory and performance:

func processInChunks(filename string, chunkSize int) ([]Employee, error) {
    file, _ := os.Open(filename)
    defer file.Close()

    reader := csv.NewReader(file)
    reader.Read()  // Skip header

    var allResults []Employee
    chunk := make([]Employee, 0, chunkSize)

    for {
        record, err := reader.Read()
        if err == io.EOF {
            // Process last chunk
            if len(chunk) > 0 {
                filtered := plygo.From(chunk).
                    Where("Department").Equals("Engineering").
                    Where("Salary").GreaterThan(80000.0).
                    Collect()
                allResults = append(allResults, filtered...)
            }
            break
        }

        id, _ := strconv.Atoi(record[0])
        salary, _ := strconv.ParseFloat(record[2], 64)

        chunk = append(chunk, Employee{
            ID: id, Department: record[1], Salary: salary,
        })

        if len(chunk) >= chunkSize {
            // Process chunk with plyGO
            filtered := plygo.From(chunk).
                Where("Department").Equals("Engineering").
                Where("Salary").GreaterThan(80000.0).
                Collect()
            allResults = append(allResults, filtered...)
            chunk = chunk[:0]  // Reset chunk
            runtime.GC()  // Force garbage collection
        }
    }

    return allResults, nil
}

runtime.GC()
printMemStats("Start")

start := time.Now()
results, _ := processInChunks("test_data.csv", 1000)
fmt.Printf("Chunked time: %v\n", time.Since(start))
fmt.Printf("Found: %d\n", len(results))

printMemStats("After Chunked")

Result

Start: 0.11 MB
Chunked time: 292.4ms
Found: 2015
After Chunked: 0.32 MB

Analysis:

• ✅ Excellent memory: Only 0.32 MB (86% less than load-all!)
• ✅ Good performance: 292ms (11x faster than load-all)
• ✅ Uses plyGO: Can leverage all plyGO features
• ✅ Scalable: Processes any size file
• ✅ Best for: Production use with large files

Performance Comparison

Strategy	Memory	Time	Scalability	plyGO Support
Load All	2.32 MB	3.37s	❌ Poor	✅ Full
Stream Indices	1.06 MB	3.52ms	✅ Excellent	⚠️ Limited
Chunked	0.32 MB	292ms	✅ Excellent	✅ Full

Recommendations:

• < 10k records: Load all - simplest approach
• 10k - 100k records: Chunked processing - best balance
• > 100k records: Stream indices first, then process matches
• Memory constrained: Always use chunked or streaming

Large Data Best Practices

Memory Optimization:

1. Use streaming when possible - read one record at a time
2. Process in chunks - batch processing with controlled memory
3. Store indices instead of full records when filtering
4. Force GC between chunks with runtime.GC()
5. Limit struct fields - only include fields you need

Performance Optimization: 6. Filter while reading - don't load unnecessary data 7. Use bufio.Scanner for line-by-line reading (even faster) 8. Parallel chunk processing - use goroutines for independent chunks 9. Index frequently filtered fields - pre-filter in streaming phase 10. Monitor memory with runtime.ReadMemStats()

Scalability:

• Chunk size matters: 1000-10000 records per chunk typically optimal
• Trade-offs: Smaller chunks = less memory, more GC overhead
• File format: Consider binary formats (Protocol Buffers, MessagePack) for very large data

Common Pitfalls

❌ Don't do this:

// Loading entire 1M record file into memory
records, _ := csv.ReadAll()  // OOM!
allData := convertAllRecords(records)  // OOM!

✅ Do this instead:

// Process in chunks
for i := 0; i < totalRecords; i += chunkSize {
    chunk := readChunk(i, chunkSize)
    processChunk(chunk)
    runtime.GC()
}

Memory leaks to avoid:

• Not closing file handles
• Accumulating results without processing
• Large slices that never get freed
• Goroutines that never complete