如何读取文件部分内容并写入剩余内容至另一文件？批量大CSV提取10%行数据

嘿，这两个都是日常开发里经常碰到的文件处理需求，我用Python来给你演示具体实现吧——毕竟它处理这类场景上手快，还能轻松应对大文件的内存问题：

1. 读取单个文件的部分内容，剩余内容写入另一文件 #

这个需求得先明确"部分内容"的定义：是按行数读取，还是按字节大小读取？我两种情况都给你写例子：

按行数读取部分内容 #

比如我们要提取前10行作为"部分内容"，剩下的全部写入新文件：

# 定义文件路径
source_path = "source_file.txt"
remaining_path = "remaining_content.txt"
# 要读取的部分内容行数
partial_line_count = 10

# 打开源文件，同时处理读取和写入
with open(source_path, 'r', encoding='utf-8') as source_f:
    # 读取前N行作为部分内容
    partial_content = [next(source_f) for _ in range(partial_line_count)]
    # 这里可以对部分内容做你需要的处理，比如打印、分析
    print("提取的部分内容：")
    for line in partial_content:
        print(line.strip())
    
    # 将剩余内容一次性写入新文件（如果文件超大，建议用分块写入）
    with open(remaining_path, 'w', encoding='utf-8') as remaining_f:
        remaining_f.writelines(source_f)

按字节大小读取部分内容 #

如果需要按字节截取部分内容（比如读取前100字节），用二进制模式处理更稳妥：

source_path = "source_file.txt"
remaining_path = "remaining_content.txt"
# 要读取的部分内容字节数
partial_byte_count = 100

with open(source_path, 'rb') as source_f:
    # 读取指定字节数的内容
    partial_content = source_f.read(partial_byte_count)
    # 转成字符串处理（注意编码要和源文件一致）
    print("提取的部分内容：")
    print(partial_content.decode('utf-8'))
    
    # 分块写入剩余内容，避免大文件占满内存
    with open(remaining_path, 'wb') as remaining_f:
        chunk_size = 4096  # 每次读4KB块
        while chunk := source_f.read(chunk_size):
            remaining_f.write(chunk)

2. 处理多个大型CSV文件，提取10%的数据写入新文件 #

处理大型CSV绝对不能一次性把文件读进内存，得逐行处理。这里分两种场景：近似10%抽样（简单高效）和精确10%抽样（保证比例准确）。

近似10%随机抽样 #

这种方式逐行读取，每一行都有10%的概率被选中，实现简单，适合对比例要求不是绝对严格的场景：

import csv
import random
from pathlib import Path

# 定义CSV文件目录和输出文件路径
csv_directory = Path("./your_csv_folder")
output_path = "sampled_10percent.csv"
sample_ratio = 0.1

# 获取目录下所有CSV文件
csv_files = list(csv_directory.glob("*.csv"))

# 打开输出文件，批量处理所有CSV
with open(output_path, 'w', newline='', encoding='utf-8') as output_f:
    csv_writer = None
    for csv_file in csv_files:
        print(f"正在处理文件：{csv_file.name}")
        with open(csv_file, 'r', newline='', encoding='utf-8') as input_f:
            csv_reader = csv.DictReader(input_f)
            # 第一次写入时生成表头
            if csv_writer is None:
                csv_writer = csv.DictWriter(output_f, fieldnames=csv_reader.fieldnames)
                csv_writer.writeheader()
            
            # 逐行随机抽样
            for row in csv_reader:
                if random.random() <= sample_ratio:
                    csv_writer.writerow(row)

精确10%抽样 #

如果需要每个文件都严格抽取10%的行数，就得先统计总行数，再随机挑选对应数量的行：

import csv
import random
from pathlib import Path

csv_directory = Path("./your_csv_folder")
output_path = "exact_10percent.csv"
sample_ratio = 0.1

csv_files = list(csv_directory.glob("*.csv"))

with open(output_path, 'w', newline='', encoding='utf-8') as output_f:
    csv_writer = None
    for csv_file in csv_files:
        print(f"正在处理文件：{csv_file.name}")
        # 先统计文件总行数（跳过表头）
        with open(csv_file, 'r', newline='', encoding='utf-8') as count_f:
            reader = csv.reader(count_f)
            next(reader)  # 跳过表头行
            total_rows = sum(1 for _ in reader)
            sample_count = int(total_rows * sample_ratio)
        
        # 随机生成要抽取的行号（数据行从第1行开始计数）
        sample_row_ids = random.sample(range(1, total_rows + 1), sample_count)
        sample_row_ids.sort()  # 排序后可以逐行遍历，避免重复跳转
        
        with open(csv_file, 'r', newline='', encoding='utf-8') as input_f:
            csv_reader = csv.DictReader(input_f)
            if csv_writer is None:
                csv_writer = csv.DictWriter(output_f, fieldnames=csv_reader.fieldnames)
                csv_writer.writeheader()
            
            current_row = 0
            row_iterator = iter(csv_reader)
            # 遍历要抽取的行号，跳过不需要的行
            for target_row in sample_row_ids:
                while current_row < target_row:
                    next(row_iterator)
                    current_row += 1
                csv_writer.writerow(next(row_iterator))
                current_row += 1

注意事项 #

• 处理大型文件时，一定要用逐行读取/分块读取，避免一次性加载整个文件导致内存溢出；
• 确保所有文件的编码一致（比如都是utf-8），不然会出现乱码；
• CSV文件如果有表头，一定要注意表头的处理，避免重复写入表头。

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