如何优化从日志文件提取数据到Pandas DataFrame的效率？

问题背景 #

处理1万行以上的日志文件时，原代码通过循环调用df.append()添加数据，因每次append都会创建新的DataFrame对象，导致整体耗时极高。以下是针对性的优化建议：

优化1：用列表收集数据，一次性生成DataFrame #

列表的append操作开销远低于DataFrame的append，先将所有有效数据存入列表，最后一次性转换为DataFrame，这是提升效率最关键的一步。

import ast
import pandas as pd
import re

infile = "./log_file.log"
data_list = []

# 逐行读取文件，避免一次性加载大文件占用过多内存
with open(infile) as f:
    for line in f:
        if "type" in line:
            # 提取日志中的字典部分
            dict_match = re.search(r'{.*}', line)
            if dict_match:
                value = ast.literal_eval(dict_match.group(0))
                # 提取时间字段
                time_str = line.split("INFO")[0].strip()
                # 提取type字段
                type_match = re.search(r"\('(.*?)'\)", line)
                if type_match:
                    value["time"] = time_str
                    value["type"] = type_match.group(1)
                    data_list.append(value)

# 一次性生成DataFrame
df = pd.DataFrame(data_list, columns=["time", "type", "value", "unit", "id", "name"])

优化2：预编译正则表达式 #

重复使用的正则表达式提前编译，避免每次匹配时重新编译，减少额外开销。

import ast
import pandas as pd
import re

# 预编译正则表达式，后续重复调用
dict_re = re.compile(r'{.*}')
type_re = re.compile(r"\('(.*?)'\)")

infile = "./log_file.log"
data_list = []

with open(infile) as f:
    for line in f:
        if "type" in line:
            dict_match = dict_re.search(line)
            if dict_match:
                value = ast.literal_eval(dict_match.group(0))
                time_str = line.split("INFO")[0].strip()
                type_match = type_re.search(line)
                if type_match:
                    value["time"] = time_str
                    value["type"] = type_match.group(1)
                    data_list.append(value)

df = pd.DataFrame(data_list, columns=["time", "type", "value", "unit", "id", "name"])

优化3：替换ast.literal_eval为字符串分割（格式固定时） #

若日志中字典部分的格式完全固定，可直接用字符串分割提取键值对，替代ast.literal_eval进一步提速。

import pandas as pd
import re

type_re = re.compile(r"\('(.*?)'\)")

infile = "./log_file.log"
data_list = []

with open(infile) as f:
    for line in f:
        if "type" in line:
            time_str = line.split("INFO")[0].strip()
            type_match = type_re.search(line)
            if not type_match:
                continue
            type_val = type_match.group(1)
            
            # 直接分割字典部分的键值对
            dict_content = line.split('{')[1].split('}')[0]
            kv_pairs = dict_content.split(', ')
            
            row = {"time": time_str, "type": type_val}
            for pair in kv_pairs:
                key, val = pair.split(': ', 1)
                # 去除字符串两端的引号
                key = key.strip("'\"")
                val = val.strip("'\"")
                # 转换数值类型
                if key == "value":
                    val = int(val)
                row[key] = val
            
            data_list.append(row)

df = pd.DataFrame(data_list, columns=["time", "type", "value", "unit", "id", "name"])

优化4：批量过滤后解析 #

先过滤出目标行再统一解析，减少无效行的处理步骤。

import ast
import pandas as pd
import re

def parse_target_line(line):
    # 提取时间
    time_str = line.split("INFO")[0].strip()
    # 提取type
    type_val = re.search(r"\('(.*?)'\)", line).group(1)
    # 提取字典内容
    dict_val = ast.literal_eval(re.search(r'{.*}', line).group(0))
    return (time_str, type_val, dict_val['value'], dict_val['unit'], dict_val['id'], dict_val['name'])

infile = "./log_file.log"

# 先过滤出包含'type'的行
with open(infile) as f:
    target_lines = [line for line in f if "type" in line]

# 批量解析并生成DataFrame
df = pd.DataFrame(
    [parse_target_line(line) for line in target_lines],
    columns=["time", "type", "value", "unit", "id", "name"]
)

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