求助：从无法用Pandas直接导入的1.6亿行CSV抽取5%随机样本

方法1：Shell命令快速抽样（推荐，性能最优） #

Linux/macOS下直接用系统级工具处理，无需加载整个文件到内存，速度最快：

精确抽取800万行（含表头） #

先统计总行数（扣除表头），再用shuf随机抽取：

# 获取总行数（减去1行表头）
total_data_lines=$(($(wc -l < input.csv) - 1))
# 先写入表头
head -n 1 input.csv > sample.csv
# 从数据行中随机抽取800万行追加到样本文件
shuf -n 8000000 <(tail -n +2 input.csv) >> sample.csv

按5%概率抽样（无需提前统计行数） #

用awk生成随机数，每行有5%概率被选中，结果接近目标样本量：

# 写入表头
head -n 1 input.csv > sample.csv
# 遍历数据行，按概率抽样
awk 'BEGIN{srand()} NR>1 {if(rand()<=0.05) print}' input.csv >> sample.csv

方法2：Python csv模块逐行处理（内存占用极低） #

适合需要自定义逻辑的场景，逐行读取文件，内存占用仅为单行列数据大小：

概率抽样版本 #

import csv
import random

random.seed(42)  # 固定随机种子，保证抽样结果可复现

with open('input.csv', 'r', newline='') as infile, open('sample.csv', 'w', newline='') as outfile:
    reader = csv.reader(infile)
    writer = csv.writer(outfile)
    
    # 写入表头
    writer.writerow(next(reader))
    
    # 逐行判断是否抽样
    for row in reader:
        if random.random() <= 0.05:
            writer.writerow(row)

精确抽取800万行版本 #

先统计总行数，再生成随机行号集合，精准抽取目标行数：

import csv
import random

# 统计数据总行数（跳过表头）
with open('input.csv', 'r') as f:
    total_data_lines = sum(1 for _ in f) - 1

# 生成要抽取的随机行号（对应数据行，从1开始计数）
target_rows = set(random.sample(range(1, total_data_lines + 1), 8000000))

with open('input.csv', 'r', newline='') as infile, open('sample.csv', 'w', newline='') as outfile:
    reader = csv.reader(infile)
    writer = csv.writer(outfile)
    
    writer.writerow(next(reader))  # 写入表头
    
    current_line = 0
    for row in reader:
        current_line += 1
        if current_line in target_rows:
            writer.writerow(row)

方法3：改进Pandas分块抽样（适配后续Pandas分析） #

如果之前用chunksize失败，大概率是没有分块写入文件导致内存溢出。试试分块处理后直接追加到文件，避免内存累积：

import pandas as pd

chunk_size = 200000  # 根据你的内存情况调整，比如20万行/块
is_first_chunk = True

for chunk in pd.read_csv('input.csv', chunksize=chunk_size):
    # 对当前块抽取5%样本
    chunk_sample = chunk.sample(frac=0.05, random_state=42)
    # 写入文件：第一块写表头，后续块跳过表头
    chunk_sample.to_csv(
        'sample.csv',
        mode='w' if is_first_chunk else 'a',
        header=is_first_chunk,
        index=False
    )
    is_first_chunk = False

这个方法的内存占用仅为单个chunk的大小，不会把所有样本都加载到内存中。

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