如何用Python高效处理大型Apache日志文件并完成匹配统计

问题背景 #

我有一个包含click_id列的CSV文件，需要用这些click_id搜索约3GB的大型Apache日志文件，匹配后提取User Agent等信息，对相似日志条目分组统计并写入新CSV。当前脚本已运行3天仍未结束，需要Python中最高效可靠的实现方式，同时解决大文件处理的内存和性能问题。

原尝试代码 #

import csv
from collections import defaultdict
from user_agents import parse

clickid_list = []
device_list = []


with open('data.csv', 'r') as file:
    reader = csv.reader(file)
    for row in reader:
        # check if click_id column is not blank or null
        if row[29] != "" and row[29] != "null" and row[29] != "click_id":
            clickid_list.append(row[29])

matched_lines_count = defaultdict(int)


def log_file_generator(filename, chunk_size=200 * 1024 * 1024):
    with open(filename, 'r') as file:
        while True:
            chunk = file.readlines(chunk_size)
            if not chunk:
                break
            yield chunk

for chunk in log_file_generator('data.log'):
    for line in chunk:
        for gclid in clickid_list:
            if gclid in line:
                string = "'" + str(line) + "'"
                user_agent = parse(string)
                device = user_agent.device.family
                device_brand = user_agent.device.brand
                device_model = user_agent.device.model
                os = user_agent.os.family
                os_version = user_agent.os.version
                browser= user_agent.browser.family
                browser_version= user_agent.browser.version

                if device in matched_lines_count:
                    matched_lines_count[device]["count"] += 1
                    print(matched_lines_count[device]["count"])
                else:
                    matched_lines_count[device] = {"count": 1, "os": os,"os_version": os_version,"browser": browser,"browser_version": browser_version,"device_brand": device_brand,"device_model": device_model}

# sort garne 
sorted_matched_lines_count = sorted(matched_lines_count.items(), key=lambda x: x[1]['count'], reverse=True)

with open("test_op.csv", "a", newline="") as file:
        writer = csv.writer(file)
        writer.writerows([["Device", "Count", "OS","OS version","Browser","Browser version","device_brand","device model"]])

        for line, count in sorted_matched_lines_count:
            # if count['count'] >= 20:
            # print(f"Matched Line: {line} | Count: {count['count']} | OS: {count['os']}")
            # write the data to a CSV file
                writer.writerow([line,count['count'],count['os'],count['os_version'],count['browser'],count['browser_version'],count['device_brand'],count['device_model']])

日志示例 #

127.0.0.1 - - [03/Nov/2022:06:50:20 +0000] "GET /access?click_id=12345678925455 HTTP/1.1" 200 39913 "-" "Mozilla/5.0 (Linux; Android 11; SM-A107F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/106.0.0.0 Mobile Safari/537.36"
127.0.0.1 - - [03/Nov/2022:06:50:22 +0000] "GET /access?click_id=123456789 HTTP/1.1" 200 39914 "-" "Mozilla/5.0 (Linux; Android 11; SM-A705FN) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Mobile Safari/537.36"

预期结果 #

生成包含设备、计数、操作系统、浏览器等分组统计信息的CSV文件。

原代码性能瓶颈分析 #

• 线性查找click_id：将click_id存储为列表，每一行日志都要遍历整个列表匹配，时间复杂度为O(N*M)（N为日志行数，M为click_id数量），这是性能慢的核心原因。
• 硬编码列索引：直接使用row[29]获取click_id，一旦CSV列顺序变化就会出错，且可读性差。
• 统计逻辑缺陷：仅以device作为统计key，会将相同设备但不同操作系统/浏览器的记录合并，导致统计结果不准确。
• 不必要的IO操作：循环中调用print，频繁的控制台输出会大幅拖慢脚本运行速度。
• User Agent解析错误：将整行日志加引号后解析，而非提取日志中真正的User Agent字段，可能导致解析结果异常。

优化思路与实现方案 #

核心优化点 #

1. 将click_id转为集合，将匹配操作从O(M)降为O(1)。
2. 使用正则表达式快速提取日志中的click_id和User Agent，避免字符串遍历。
3. 使用复合键（设备、系统、浏览器等组合）作为统计维度，保证统计准确性。
4. 用csv.DictReader读取CSV，避免硬编码列索引，提升代码健壮性。
5. 逐行处理日志（分块读取但逐行解析），降低内存占用。
6. 移除不必要的print操作，减少IO开销。

完整优化代码 #

import csv
import re
from collections import defaultdict
from user_agents import parse

# 1. 读取click_id并转为集合（O(1)查找）
click_id_set = set()
with open('data.csv', 'r', encoding='utf-8') as csv_file:
    reader = csv.DictReader(csv_file)
    for row in reader:
        click_id = row.get('click_id', '').strip()
        if click_id and click_id != 'null':
            click_id_set.add(click_id)

# 2. 预编译正则表达式，提取日志中的click_id和User Agent
log_pattern = re.compile(
    r'GET /access\?click_id=([^\s]+).*?"([^"]+)"$'
)

# 3. 统计字典：复合键保证统计维度准确
stats = defaultdict(int)

# 4. 逐行处理日志文件，避免内存过载
with open('data.log', 'r', encoding='utf-8') as log_file:
    for line in log_file:
        line = line.strip()
        if not line:
            continue
        
        match = log_pattern.search(line)
        if not match:
            continue
        
        extracted_click_id, user_agent_str = match.groups()
        if extracted_click_id not in click_id_set:
            continue
        
        # 解析User Agent
        ua = parse(user_agent_str)
        # 构建复合统计键
        stat_key = (
            ua.device.family,
            ua.os.family,
            '.'.join(ua.os.version),
            ua.browser.family,
            '.'.join(ua.browser.version),
            ua.device.brand or 'Unknown',
            ua.device.model or 'Unknown'
        )
        stats[stat_key] += 1

# 5. 排序并写入结果CSV
sorted_stats = sorted(stats.items(), key=lambda x: x[1], reverse=True)

with open('result_stats.csv', 'w', newline='', encoding='utf-8') as out_file:
    writer = csv.writer(out_file)
    # 写入表头
    writer.writerow([
        "Device", "OS", "OS Version", "Browser", 
        "Browser Version", "Device Brand", "Device Model", "Count"
    ])
    # 写入统计数据
    for key, count in sorted_stats:
        writer.writerow(list(key) + [count])

关键优化说明 #

• 集合查找：集合的成员检测基于哈希表实现，时间复杂度O(1)，相比列表的O(M)，当click_id数量较大时性能提升极其明显。
• 正则预编译：提前编译正则表达式避免重复编译开销，同时精准提取目标字段，避免无效的字符串遍历。
• 复合统计键：确保相同设备但不同系统/浏览器的记录不会被错误合并，统计结果更准确。
• 逐行处理日志：无需一次性加载整个3GB日志到内存，内存占用始终保持在较低水平。
• 健壮的CSV读取：使用DictReader通过列名获取click_id，避免硬编码索引导致的错误。

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