Python中两种不同格式大文件的高效对比最优方案及代码优化咨询

Great question! Your current approach is hitting performance bottlenecks because it re-reads and re-scans the entire file 2 every single time you look up an ID from file 1. With 20k+ lines in each file, that's an O(N*M) time complexity—definitely going to get slow as you process more lines. Let's fix this using the key hint you provided: file 2's IDs are sorted in ascending order.

Why Your Current Code Is Slow #

Let's recap the core issue:

• Every time you find a Type=New entry in file 1, you open file 2 from scratch and scan every line until you find a matching ID.
• This means for 20k entries in file 1, you could end up scanning file 2 20k times total—huge waste of IO and processing power.

The Optimal Approach: Leverage Sorted IDs #

Since file 2's IDs are strictly increasing, we only need to scan file 2 once. Here are two efficient implementations depending on your needs:

Option 1: Precollect Target IDs + Single Pass of File 2 #

This works best if you just need to count matches or track which IDs exist in both files:

import json

def compare_large_files(file1_path, file2_path):
    # Step 1: Extract all `Type=New` IDs from file 1 and store in a set
    target_ids = set()
    with open(file1_path, "r") as f1:
        for line in f1:
            line = line.strip()
            if not line:
                continue
            # Parse file1's custom format (handle cases where Desc might contain "=")
            entry = {}
            for part in line.split(" "):
                key, value = part.split("=", 1)
                entry[key] = value
            if entry.get("Type") == "New":
                target_ids.add(entry["Id"])
    
    # Step 2: Scan file2 ONCE to find matches
    match_count = 0
    with open(file2_path, "r") as f2:
        for line in f2:
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue  # Skip invalid JSON lines
            if entry.get("Type") == "New":
                current_id = entry["Id"]
                if current_id in target_ids:
                    match_count += 1
                    target_ids.remove(current_id)  # Avoid duplicate matches
                    if not target_ids:
                        break  # Exit early if all targets are found
                # Early exit: since IDs are sorted, no need to check further if we pass the largest target ID
                if target_ids and current_id > max(target_ids):
                    break
    return match_count

Option 2: Two-Pointer Technique (Even Faster for Sorted Targets) #

If we sort the target IDs from file 1 first, we can use a two-pointer approach to sync traversal of both files—no set lookups needed:

import json

def compare_with_two_pointers(file1_path, file2_path):
    # Step 1: Extract and sort `Type=New` IDs from file1
    target_ids = []
    with open(file1_path, "r") as f1:
        for line in f1:
            line = line.strip()
            if not line:
                continue
            entry = {}
            for part in line.split(" "):
                key, value = part.split("=", 1)
                entry[key] = value
            if entry.get("Type") == "New":
                target_ids.append(entry["Id"])
    
    target_ids.sort()
    target_count = len(target_ids)
    if target_count == 0:
        return 0
    
    # Step 2: Two-pointer traversal
    match_count = 0
    target_ptr = 0  # Tracks which ID we're looking for in file1's sorted list
    with open(file2_path, "r") as f2:
        for line in f2:
            line = line.strip()
            if not line:
                continue
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue
            if entry.get("Type") == "New":
                current_id = entry["Id"]
                # Move pointer forward until we find a matching or larger ID
                while target_ptr < target_count and current_id > target_ids[target_ptr]:
                    target_ptr += 1
                # Check for match
                if target_ptr < target_count and current_id == target_ids[target_ptr]:
                    match_count += 1
                    target_ptr += 1
                    if target_ptr >= target_count:
                        break  # All matches found, exit early
                # Early exit if we've passed the last target ID
                if current_id > target_ids[-1]:
                    break
    return match_count

Bonus Optimization Tips #

• Avoid readlines(): Your original code uses readlines() which loads the entire file into memory. For large files, iterating line-by-line (for line in file) is far more memory-efficient.
• Handle edge cases: Add error handling for invalid lines (malformed JSON, broken custom format) to prevent crashes.
• For extremely large file2: If file2 is massive (100k+ lines), you could precompute a list of byte offsets for all Type=New entries, then use binary search to jump directly to potential matches. This adds some setup overhead but speeds up lookups significantly.

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