大输出场景下multiprocessing.Process结合Queue的优化方案咨询

Let's tackle your performance issues step by step—your goal of parallelizing CPU-heavy line processing makes total sense, but the way you're handling inter-process communication and some code inefficiencies are holding you back.

First, Fix Critical Bugs in Your Code #

Before diving into optimizations, let's squash a couple of bugs that'll break your results:

• In fLine(), you're using rOutput.split(" ") instead of fOutput.split(" ")—this typo will mix up your "f" record data with "r" record data.
• In your main block, you have typos: process.append(...) should be processes.append(...), and the final loop uses for p in process: instead of for p in processes:.

Why Your Current Multiprocessing Approaches Are Slow/Hanging #

1. Queue Overhead with Large Dictionaries: When you pass a full defaultdict through a multiprocessing.Queue, Python has to serialize (pickle) the entire dictionary, send it across process boundaries, then deserialize it. For large datasets, this is extremely slow and can cause hangs as the queue tries to handle massive amounts of data.
2. Pool Underperformance: If you tried Pool and saw slower speeds, it's likely because the overhead of spawning processes and coordinating them outweighed the parallel gains—especially if your split files were too small, leading to frequent context switching.

Optimized Solutions for Faster Processing #

1. Avoid Inter-Process Dictionary Transfers: Use Temporary Files Instead

Instead of sending entire dictionaries back to the main process, have each worker process write its results to a temporary file (using a compact format like pickle). The main process can then read all these temp files and merge the results. This eliminates the huge serialization/deserialization overhead.

Example modification to your mainFunction:

import pickle
import tempfile
import os
from collections import defaultdict
import gzip

def mainFunction(thisFile):
    resultDict = defaultdict(list)
    # Use a larger buffer for gzip to reduce IO overhead
    with gzip.open(thisFile, 'rb', buffer_size=10*1024*1024) as f:
        for lines in f:
            line = lines.decode().split()  # split() handles all whitespace faster than split(" ")
            record = line[0]
            if record == "f":
                id_val = line[7]
                output_str = ' '.join(line[4:6])  # More efficient than string concatenation
                resultDict[id_val].extend(output_str.split())
            else:
                id_val = line[2]
                output_str = ' '.join(line[4:6])
                resultDict[id_val].extend(output_str.split())
    
    # Write result to a temporary pickle file
    temp_file = tempfile.NamedTemporaryFile(delete=False)
    pickle.dump(resultDict, temp_file)
    temp_file.close()
    return temp_file.name

Then in your main block, use Pool to process files and collect temp file paths:

import multiprocessing

if __name__ == '__main__':
    finalDict = defaultdict(list)
    files = ["f1.log.gz", "f2.log.gz"]  # Your full list of files
    
    # Use Pool with a process count matching your CPU cores
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        temp_files = pool.map(mainFunction, files)
    
    # Merge all temp files into finalDict
    for temp_file in temp_files:
        with open(temp_file, 'rb') as f:
            worker_dict = pickle.load(f)
            for id_val, items in worker_dict.items():
                finalDict[id_val].extend(items)
        # Clean up temp file
        os.unlink(temp_file)
    
    # Use finalDict as needed
    print(dict(finalDict))

2. Optimize Line Processing Speed

• Buffer Gzip Reads: Adding a large buffer_size to gzip.open reduces the number of IO operations, which is critical for large files.
• Faster String Handling: Replace split(" ") with split() (splits on any whitespace, faster) and use ' '.join(line[4:6]) instead of line[4] + " " + line[5] (more efficient concatenation).
• Cut Function Call Overhead: For extremely tight loops, moving the logic from rLine() and fLine() directly into the loop can save time—function calls add small overhead that accumulates over millions of lines.

3. Optimize File Splitting

Make sure your split files are large enough—aim for at least 200-500MB per file (uncompressed). If your split files are too small, the overhead of spawning and managing processes will eat into any parallel gains.

4. Monitor CPU Utilization

Use tools like htop to check if your CPU is fully utilized. If it's not, your bottleneck might be IO (try faster storage or larger buffers) or your code still has unaddressed inefficiencies.

内容的提问来源于stack exchange，提问作者Praniti Gupta