日志统计提取技术咨询：从指定日志中提取指标的实现方案

需求说明 #

需要从指定格式的日志文件中提取两类统计指标：

• 单迭代数据提取：以data received processing started为起始标记、time taken为结束标记，提取每个迭代的总数据量、转换时间、处理完成时间、总耗时
• 多迭代统计分析：计算所有迭代中数据量的最小值、最大值、平均值，并对应获取这些极值所在迭代的转换时间、处理时间和总耗时

用户已尝试将日志从CSV转JSON，但后续提取指标遇到瓶颈，以下是直接解析日志并完成统计的优化方案。

日志样本 #

2023-03-09 18:55:56,INFO,capturing: DataIngestion
2023-03-09 18:55:57,INFO,waiting to get data
2023-03-09 18:56:58,INFO,time started,2023-03-09 18:56:57
2023-03-09 18:56:59,INFO,data received processing started
2023-03-09 18:56:59,INFO,data to convert, 23000
2023-03-09 18:56:00,INFO,covert json,2023-03-09 18:56:57
2023-03-09 18:57:00,INFO,convesion time,0 days 00:01:00
2023-03-09 18:57:58,INFO,process data,2023-03-09 18:57:58
2023-03-09 18:59:01,INFO,process completed,0 days 00:02:03
2023-03-09 18:59:10,INFO,time taken,0 days 00:00:09
2023-03-09 18:59:02,INFO,waiting to get data
2023-03-09 18:59:03,INFO,time started,2023-03-09 18:59:03
2023-03-09 18:59:59,INFO,data received processing started
2023-03-09 18:59:59,INFO,data to convert,30000
2023-03-09 19:00:01,INFO,covert json,2023-03-09 19:00:01
2023-03-09 19:01:31,INFO,convesion time,0 days 00:01:30
2023-03-09 19:01:32,INFO,process data,2023-03-09 19:01:32
2023-03-09 19:04:30,INFO,process completed,0 days 00:03:28
2023-03-09 19:04:31,INFO,time taken,0 days 00:03:31

实现代码 #

直接解析日志并提取迭代数据 #

无需先转JSON，直接逐行解析日志更高效，代码如下：

def parse_log(file_path):
    iterations = []
    current_iter = {}
    
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            # 分割字段，仅拆分前3个逗号，避免日志内容中的逗号干扰
            parts = line.split(',', 3)
            log_time, level, log_content = parts[0], parts[1], parts[2].strip()
            extra_data = parts[3].strip() if len(parts) > 3 else ""
            
            # 标记迭代开始，初始化临时存储字典
            if log_content == "data received processing started":
                current_iter = {}
            # 提取当前迭代的转换数据量
            elif log_content == "data to convert":
                current_iter["data_volume"] = int(extra_data)
            # 提取转换耗时（注意日志拼写错误：convesion）
            elif log_content == "convesion time":
                current_iter["conversion_time"] = extra_data
            # 提取处理完成耗时
            elif log_content == "process completed":
                current_iter["process_time"] = extra_data
            # 标记迭代结束，将当前迭代数据存入列表
            elif log_content == "time taken":
                current_iter["total_time"] = extra_data
                iterations.append(current_iter.copy())
    
    return iterations

# 解析日志文件
all_iterations = parse_log('test1.log')

# 输出各迭代详细数据
print("=== 各迭代详细数据 ===")
for idx, iter_data in enumerate(all_iterations, 1):
    print(f"迭代 {idx}:")
    print(f"  数据量: {iter_data['data_volume']}")
    print(f"  转换时间: {iter_data['conversion_time']}")
    print(f"  处理完成时间: {iter_data['process_time']}")
    print(f"  总耗时: {iter_data['total_time']}\n")

# 统计数据量极值及对应迭代信息
print("=== 数据量统计分析 ===")
if not all_iterations:
    print("无有效迭代数据")
else:
    # 提取所有迭代的 data_volume 用于统计
    data_volumes = [iter["data_volume"] for iter in all_iterations]
    avg_volume = sum(data_volumes) / len(data_volumes)
    
    # 找到数据量最小值对应的迭代
    min_vol = min(data_volumes)
    min_iter = next(iter for iter in all_iterations if iter["data_volume"] == min_vol)
    # 找到数据量最大值对应的迭代
    max_vol = max(data_volumes)
    max_iter = next(iter for iter in all_iterations if iter["data_volume"] == max_vol)
    
    print(f"数据量平均值: {avg_volume:.2f}")
    print(f"\n数据量最小值: {min_vol}")
    print(f"  对应迭代转换时间: {min_iter['conversion_time']}")
    print(f"  对应迭代处理时间: {min_iter['process_time']}")
    print(f"  对应迭代总耗时: {min_iter['total_time']}")
    print(f"\n数据量最大值: {max_vol}")
    print(f"  对应迭代转换时间: {max_iter['conversion_time']}")
    print(f"  对应迭代处理时间: {max_iter['process_time']}")
    print(f"  对应迭代总耗时: {max_iter['total_time']}")

代码说明 #

1. 日志解析逻辑：
   • 逐行读取日志，用split(',', 3)分割字段，避免日志内容中的逗号干扰拆分结果
   • 用current_iter临时存储当前迭代的所有数据，遇到迭代起始标记时初始化字典，遇到结束标记时将数据存入迭代列表
2. 数据提取：
   • 捕获data to convert的数值作为数据量
   • 捕获convesion time（兼容日志拼写错误）作为转换耗时
   • 捕获process completed的时长作为处理完成耗时
   • 捕获time taken作为迭代总耗时
3. 统计分析：
   • 计算数据量平均值，筛选出最小值、最大值对应的迭代，输出对应时间指标

运行结果示例 #

=== 各迭代详细数据 ===
迭代 1:
  数据量: 23000
  转换时间: 0 days 00:01:00
  处理完成时间: 0 days 00:02:03
  总耗时: 0 days 00:00:09

迭代 2:
  数据量: 30000
  转换时间: 0 days 00:01:30
  处理完成时间: 0 days 00:03:28
  总耗时: 0 days 00:03:31

=== 数据量统计分析 ===
数据量平均值: 26500.00

数据量最小值: 23000
  对应迭代转换时间: 0 days 00:01:00
  对应迭代处理时间: 0 days 00:02:03
  对应迭代总耗时: 0 days 00:00:09

数据量最大值: 30000
  对应迭代转换时间: 0 days 00:01:30
  对应迭代处理时间: 0 days 00:03:28
  对应迭代总耗时: 0 days 00:03:31

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