使用Python预处理后将30GB以上超大TXT文件转为CSV的最高效方法

处理超大型TXT转CSV的可行方案 #

核心思路是放弃全量读取文件的逻辑，改用流式逐块处理，全程内存占用稳定在几十MB级别，完全支持几十上百GB的文件处理。

方案1：Python原生流式处理（无额外依赖，兼容性最好） #

直接逐块读取文件，处理完就写入CSV，全程不加载全量数据到内存：

import csv

CHUNK_SIZE = 1024 * 1024 * 10  # 单次读取10MB，可根据剩余内存调整大小
input_path = "myfile.txt"
output_path = "myfile.csv"
remainder = ""  # 存储上一个读取块末尾不完整的记录

with open(input_path, 'r', encoding='utf-8') as in_f, open(output_path, 'w', encoding='utf-8', newline='') as out_f:
    writer = csv.writer(out_f)
    # 写入表头
    writer.writerow(['a', 'b', 'c', 'd'])
    
    while True:
        chunk = in_f.read(CHUNK_SIZE)
        # 所有内容读取完成，处理最后剩余的不完整记录
        if not chunk:
            if remainder:
                cleaned = remainder.replace("'", "").rstrip("#@#@#")
                if cleaned:
                    writer.writerow(cleaned.split("~"))
            break
        # 拼接上一轮剩余的不完整内容
        full_content = remainder + chunk
        # 按记录分隔符拆分
        records = full_content.split("#@#@#")
        # 最后一个元素是当前块末尾不完整的记录，留到下一轮处理
        remainder = records.pop()
        # 处理当前完整记录并写入
        for record in records:
            cleaned = record.replace("'", "")
            if not cleaned:
                continue
            fields = cleaned.split("~")
            # 可选校验字段数，避免脏数据报错
            if len(fields) == 4:
                writer.writerow(fields)

方案2：Pandas分块处理（适合习惯Pandas操作的场景） #

利用Pandas内置的分块读取能力，无需自己处理IO逻辑：

import pandas as pd

CHUNK_SIZE = 10**6  # 单次处理100万行，可根据内存调整
input_path = "myfile.txt"
output_path = "myfile.csv"
first_write = True

for chunk in pd.read_csv(
    input_path,
    sep="~",
    chunksize=CHUNK_SIZE,
    names=['raw_a', 'raw_b', 'raw_c', 'raw_d'],
    lineterminator="#"
):
    # 清理多余符号
    chunk['a'] = chunk['raw_a'].str.replace("'", "", regex=False)
    chunk['b'] = chunk['raw_b'].str.replace("'", "", regex=False)
    chunk['c'] = chunk['raw_c'].str.replace("'", "", regex=False)
    chunk['d'] = chunk['raw_d'].str.replace("'", "", regex=False).str.replace("@#@", "", regex=False)
    # 只保留有效列
    df = chunk[['a', 'b', 'c', 'd']].dropna()
    # 追加写入CSV，仅第一个块写入表头
    df.to_csv(output_path, mode='a', index=False, header=first_write)
    first_write = False

额外性能优化建议 #

• 可以根据设备剩余内存适当调大CHUNK_SIZE，减少IO次数提升速度
• 如果你使用Linux/macOS环境，直接用系统命令行工具处理速度比Python快3~5倍，命令参考：

sed "s/'//g;s/#@#@#/\n/g" myfile.txt | awk 'BEGIN{FS="~";OFS=","}{print $1,$2,$3,$4}' > myfile.csv

内容的提问来源于stack exchange，提问作者Aquiles Páez