如何高效拆分含多类数据的大型CSV文件至不同DataFrame？

核心思路 #

避开逐行处理的低效，利用pandas的批量读取能力，结合行结构特征（列数、标识字符串）快速拆分数据，同时兼顾内存占用。

方案1：先识别分界点再批量加载 #

如果文件里初始状态行、事件行、日志行有明确区分规则（比如初始行带[INIT]标识，事件行固定2列，日志行≥20列），先快速扫一遍文件定位各部分的行索引，再用read_csv的参数直接加载对应区块：

import pandas as pd

file_path = "target_file.log"

# 快速扫描定位各数据块的行边界
init_rows = []
event_start_idx = None
log_start_idx = None

# 用带缓冲的文件读取提升扫描速度
with open(file_path, 'r', encoding='utf-8', buffering=1024*1024) as f:
    for line_idx, line in enumerate(f):
        stripped_line = line.strip()
        if not stripped_line:
            continue
        # 按实际规则判断初始状态行
        if stripped_line.startswith("[INIT]"):
            init_rows.append(line_idx)
        elif event_start_idx is None:
            # 判断是否为事件行（分割后列数为2）
            col_count = len(stripped_line.split(','))
            if col_count == 2:
                if event_start_idx is None:
                    event_start_idx = line_idx
            elif col_count >= 20:
                # 定位到日志行起始点，停止扫描
                log_start_idx = line_idx
                break

# 加载初始状态+事件数据
init_event_df = pd.read_csv(
    file_path,
    skiprows=lambda x: x >= log_start_idx if log_start_idx else False,
    header=None  # 根据实际文件是否有表头调整
)

# 加载日志数据
log_df = pd.read_csv(
    file_path,
    skiprows=log_start_idx,
    header=None
)

方案2：分块读取+批量筛选 #

如果只有列数差异（事件行2列、日志行≥20列）作为区分依据，用chunksize分块读取文件，每块内按列数筛选拆分：

import pandas as pd

file_path = "target_file.log"
chunk_size = 2000  # 根据内存情况调整，越大读取效率越高

init_event_chunks = []
log_chunks = []

# 分块读取并筛选
for chunk in pd.read_csv(file_path, chunksize=chunk_size, header=None):
    # 按非空列数判断行类型，可根据实际格式调整规则
    event_chunk = chunk[chunk.notna().sum(axis=1) == 2]
    log_chunk = chunk[chunk.notna().sum(axis=1) >= 20]
    init_event_chunks.append(event_chunk)
    log_chunks.append(log_chunk)

# 合并为最终DataFrame
init_event_df = pd.concat(init_event_chunks, ignore_index=True)
log_df = pd.concat(log_chunks, ignore_index=True)

额外优化点 #

• 固定文件编码：指定encoding参数，避免pandas自动编码检测的耗时
• 并行处理多文件：用multiprocessing启动多进程，每个进程处理一个文件，最后合并结果
• 跳过空行：读取时添加skip_blank_lines=True参数，减少无效行处理

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