在Julia中高效合并65000个CSV文件并实现基础统计的优化方案咨询

Great question—dealing with tens of thousands of CSV files in Julia can feel like fighting a memory hog when you try to cram everything into one giant DataFrame. Let’s break down the best strategies to handle this efficiently, depending on exactly what kind of stats you need to run:

1. Skip Full Merging Entirely: Calculate Stats Per File First #

This is the most efficient approach for basic statistics (means, medians, sums, min/max). Instead of merging all data, compute the stats you need for each individual file, then aggregate those results. You’ll never hold more than one small file in memory at a time.

Example Code #

using CSV, DataFrames, Statistics

csv_dir_tmax = cd(readdir, "C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax")

# Initialize a DataFrame to store per-file statistics
file_stats = DataFrame(
    filename = String[],
    row_count = Int[],
    tmax_mean = Float64[],
    tmax_median = Float64[],
    tmax_min = Float64[],
    tmax_max = Float64[]
)

for (idx, fname) in enumerate(csv_dir_tmax)
    println("Processing file $idx: $fname")
    file_path = joinpath("C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax", fname)
    df = CSV.read(file_path, DataFrame)
    
    # Compute stats for this file and add to our results
    push!(file_stats, (
        fname,
        nrow(df),
        mean(df.TMAX),
        median(df.TMAX),
        minimum(df.TMAX),
        maximum(df.TMAX)
    ))
end

# Example: Calculate a weighted overall mean across all files
overall_tmax_mean = sum(file_stats.tmax_mean .* file_stats.row_count) / sum(file_stats.row_count)

Why This Works #

• Memory usage stays tiny: Each file is processed, stats are extracted, then the original file data is garbage-collected immediately.
• Speed is maximized: No time wasted copying large chunks of data between DataFrames.

2. Incremental Aggregation for Cross-File Groups #

If you need stats grouped by a shared column (like Date across all files), don’t merge everything first. Instead, update running totals for each group as you process each file.

Example Code (Date-Based Stats) #

using CSV, DataFrames

csv_dir_tmax = cd(readdir, "C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax")

# Use a dictionary to track running totals per date: key=Date, value=(sum_TMAX, count_TMAX)
date_totals = Dict{Date, Tuple{Float64, Int}}()

for (idx, fname) in enumerate(csv_dir_tmax)
    println("Processing file $idx: $fname")
    file_path = joinpath("C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax", fname)
    df = CSV.read(file_path, DataFrame)
    
    for row in eachrow(df)
        dt = row.Date
        tmax_val = row.TMAX
        
        if haskey(date_totals, dt)
            current_sum, current_count = date_totals[dt]
            date_totals[dt] = (current_sum + tmax_val, current_count + 1)
        else
            date_totals[dt] = (tmax_val, 1)
        end
    end
end

# Convert the dictionary to a DataFrame for final analysis
date_stats = DataFrame(
    Date = collect(keys(date_totals)),
    total_TMAX = [val[1] for val in values(date_totals)],
    sample_count = [val[2] for val in values(date_totals)]
)

date_stats.TMAX_mean = date_stats.total_TMAX ./ date_stats.sample_count

Why This Works #

• Memory usage depends on the number of unique groups (e.g., unique dates), not the total number of rows across all files.
• You avoid loading and merging millions of rows into a single DataFrame.

3. Your Chunking Idea: Make It Stat-First, Not Merge-First #

If you must work with chunks (e.g., for more complex analyses that can’t be done per-file), modify your approach to compute stats on each chunk before moving to the next. This way, you never store full chunks of raw data long-term.

Example Code #

using CSV, DataFrames, Statistics

csv_dir_tmax = cd(readdir, "C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax")
chunk_size = 20000  # Adjust based on your available memory

# Track overall stats across all chunks
overall_sum = 0.0
overall_count = 0

for chunk_start in 1:chunk_size:length(csv_dir_tmax)
    chunk_end = min(chunk_start + chunk_size - 1, length(csv_dir_tmax))
    println("Processing chunk: files $chunk_start to $chunk_end")
    
    # Compute stats for this chunk
    chunk_sum = 0.0
    chunk_count = 0
    for fname in csv_dir_tmax[chunk_start:chunk_end]
        file_path = joinpath("C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax", fname)
        df = CSV.read(file_path, DataFrame)
        chunk_sum += sum(df.TMAX)
        chunk_count += nrow(df)
    end
    
    # Update overall stats
    overall_sum += chunk_sum
    overall_count += chunk_count
    
    # Optional: Save chunk stats to CSV for record-keeping
    CSV.write("chunk_stats_$chunk_start.csv", DataFrame(chunk_sum=[chunk_sum], chunk_count=[chunk_count]))
end

# Calculate final overall statistics
overall_tmax_mean = overall_sum / overall_count

4. Pre-Initializing a DataFrame (If You Must Merge) #

Your idea to pre-initialize a DataFrame is better than repeated vcat() calls—vcat() creates a new DataFrame every time, which wastes time copying data. Here’s how to implement it correctly:

Example Code #

using CSV, DataFrames

csv_dir_tmax = cd(readdir, "C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax")

# First pass: Calculate total number of rows across all files (fast, no full load)
total_rows = 0
for fname in csv_dir_tmax
    file_path = joinpath("C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax", fname)
    total_rows += CSV.File(file_path; header=true) |> length
end

# Pre-initialize the final DataFrame with exact size and column types
tmax_all = DataFrame(
    Date = Vector{Date}(undef, total_rows),
    TMAX = Vector{Float64}(undef, total_rows)
)

# Second pass: Fill the pre-allocated DataFrame
current_row = 1
for (idx, fname) in enumerate(csv_dir_tmax)
    println("Processing file $idx: $fname")
    file_path = joinpath("C:/Users/andrew.bannerman/Desktop/Julia/scripts/GHCN data/ghcnd_all_csv/tmax", fname)
    df = CSV.read(file_path, DataFrame)
    num_rows = nrow(df)
    
    # Copy data directly into the pre-allocated slots
    tmax_all[current_row:current_row+num_rows-1, :] = df
    current_row += num_rows
end

Note #

Only use this if your total dataset fits comfortably in memory. For 65k files, this is only feasible if each file is very small.

Final Recommendation #

For basic statistical analysis, go with either per-file stats or incremental group aggregation. These methods eliminate the need to merge massive datasets entirely, which is the root cause of your slowdown and memory issues.

内容的提问来源于stack exchange，提问作者Andrew Bannerman