简介:面向深度学习初学者的CSV数据预处理工具包,内含4个Python脚本,压缩包仅2KB,适合有一定Python基础但缺乏数据规范经验的开发者研读。脚本围绕pandas库展开,覆盖缺失值填充、异常值处理、Z-score与Min-Max标准化、特征相关性分析、one-hot编码与train_test_split数据集划分等常见环节,并演示如何借助numpy数组将预处理结果转换为TensorFlow或PyTorch所需的张量格式。每个脚本相对独立,既可以直接运行查看效果,也方便抽取其中某个函数集成到自己的数据流水线中,减少重复试错成本。已有328人学习下载,适合正在准备结构化数据任务的开发者参考,也可作为深度学习数据预处理阶段的配套示例,帮助理解规范化流程对模型泛化能力的影响,从而把更多精力放在模型设计上。
1. 一个 zip 包里的深度学习项目:CSV 文件为什么值得单独处理
“处理csv文件深度学习.zip”这类命名,在网盘转存、课程资源和聊天传包里很常见。里面通常是一份 csv 数据、一段训练脚本和一个说明文档。收到包的人最容易被“深度学习”四个字吸引,急着解压、急着跑训练,结果在数据读取阶段就被编码、分隔符、类型推断挨个绊倒。csv 是深度学习中表格场景最普遍的输入格式,但它不是从文件一步到位变成模型输入的,中间有解压校验、编码探测、类型压缩、清洗划分好几道工序。任何一道出错,模型再先进也是在拿脏数据训练,后面调参全是白费。这篇笔记面向第一次把手头 csv 数据接进深度学习项目的同学,也适合被验证集泄漏、标签错位这类问题坑过的人。核心就一句话:把 csv 处理流程写得可复现,你的深度学习项目就成功了一半。
2. 解压 zip 并建立项目目录:先让代码和数据各归其位
2.1 用 Python 安全解压 zip:路径穿越和中文编码怎么处理
拿到 zip 包先用命令行解压只能算应急操作。深度学习项目要反复复现、重跑,我一般会把解压动作也写成脚本。你会在这种数据处理包里遇到两个问题:一是 zip 内有一层多余的嵌套文件夹,二是 Windows 下压缩、Linux 下解压后中文 csv 文件名变成乱码。先用一段安全解压代码把这两件事处理掉:
import zipfile from pathlib import Path def safe_extract(zip_path: str, target_dir: str) -> None: target = Path(target_dir).resolve() if not target.exists(): target.mkdir(parents=True) with zipfile.ZipFile(zip_path) as zf: members = zf.infolist() for member in members: # 目标路径必须落在 target 之内,防止 ../ 路径穿越 dest_path = (target / member.filename).resolve() if not str(dest_path).startswith(str(target)): raise ValueError(f"非法路径: {member.filename}") # 处理 Windows 压缩环境下常见的中文编码错位 name = member.filename.encode('cp437').decode('gbk', errors='ignore') member.filename = name zf.extractall(target) print(f"解压完成,共 {len(members)} 个文件") if __name__ == "__main__": safe_extract("处理csv文件深度学习.zip", "project")为什么先resolve()再判断 startswith:因为 zip 条目可能是../data.csv或a/../../b.csv,这在混合操作系统环境下很常见。不校验直接 extract,文件会写到 target 目录之外,轻则目录结构混乱,重则覆盖服务器上的其他文件。中文乱码这一步的原理是基于 zip 规范里的文件名默认用 cp437 编码,Windows 中文压缩包把 gbk 文件名写进了 cp437 字段,Python 按 cp437 读出乱码,这里把它还原成 gbk。有些包其实是 utf-8 文件名,这时改会报错,可以先解压出来看乱码形态再决定用哪种解码。
如果你不想在解压时改文件名,也可以解压后单独跑一个重命名脚本。但我建议还是解压时就处理完,后面 CSV 路径引用一致,不用对着乱码目录反复人工识别。这个安全解压函数也可以作为项目通用工具保留下来,下次收到乱七八糟的包直接调用,省得每次重写。
2.2 zip 伪加密与密码保护:遇到打不开的包先别删
第二个常见情况是 zip 包带密码。先说一个值得单独拎出来的坑——zip 伪加密。很多数据包的制作者用“没真正加密但设置了加密标志”的方式打包,或者早期压缩工具把 general purpose bit flag 第 0 位误标为 1,结果大家用unzip data.zip时提示需要密码。数据本身并没加密,只是标志位在说谎。这种伪加密可以写脚本修复:
def fix_fake_encryption(zip_path: str) -> str: """ 修复伪加密:把本地文件头和中央目录中的 general purpose flag 第 0 位清零。仅限数据未真正加密的场景。 """ dst = zip_path.replace('.zip', '_fixed.zip') with open(zip_path, 'rb') as f: data = bytearray(f.read()) n = len(data) fixed = 0 i = 0 # 扫描每个 entry 的文件头标识 # 本地文件头 PK\x03\x04 的标志位在偏移 6 # 中央目录头 PK\x01\x02 的标志位在偏移 8 while i < n - 4: if data[i:i+4] == b'PK\x03\x04': data[i + 6] &= 0xFE fixed += 1 i += 4 elif data[i:i+4] == b'PK\x01\x02': data[i + 8] &= 0xFE fixed += 1 i += 4 else: i += 1 with open(dst, 'wb') as f: f.write(data) print(f"修复了 {fixed} 处加密标志位,输出 {dst}") return dst这里的逻辑是定位 zip 二进制里的文件头标识,把标志位按位清零。只改动加密位,不动 CRC 与中央目录偏移,所以不会影响其他数据。修复后再用zipfile.ZipFile打开就不要求密码了。但注意,如果 zip 是真的用 ZipCrypto 或 AES 加密过,清掉位标记只会得到一堆损坏文件。真正加密的包只有一条路——找制作者要密码。拿到密码后正规做法是:
import zipfile with zipfile.ZipFile('data_encrypted.zip') as zf: zf.setpassword(b'your_password_here') zf.extractall('project')不要折腾暴力破解工具,时间成本远高于重新要一次密码,而且这也不是工程上该推广的做法。伪加密修复只适合明确知道数据未加密、只是标志位出错的情形。
2.3 解压后的文件清单与磁盘占用检查
解压完别急着pandas 读 csv,先把解压结果盘一遍。深度学习训练中途磁盘写满会让 checkpoint 落盘失败,之前的 epoch 全部作废,这是最不值当的翻车。我用一段 bash 命令快速出清单:
cd project find . -type f -name "*.csv" -exec ls -lh {} \; > csv_files.txt du -sh ./* cat csv_files.txtfind把项目里所有 csv 文件找出来并按行写入 csv_files.txt,du -sh看每个子目录占用。如果发现某一份 csv 占用异常大,先确认它是你需要的全量数据,而不是测试时误写的中间结果。对照 zip 包原信息检查文件数量,若差了关键文件,基本就是压缩包不完整,趁早重新下载或复制,别等训练到一半才报文件不存在。
3. 用 pandas 读入 CSV:编码、分隔符和 dtype 三个参数定生死
3.1 读入前先探测:编码、分隔符和表头检测
pandas 导入 csv 文件是日常操作,但真正动手时第一个撞上的多半不是内存问题,而是编码和分隔符。Excel 另存出来的 csv 经常是 gbk 编码,Linux 下生成的大多是 utf-8;如果 csv 开头带 BOM,按utf-8读取会把第一列列名读成\ufeffuser_id这种带脏前缀的字段。分隔符更不用提,中文数据用全角逗号、汇总文本用制表符都很常见。我习惯在读入前先用一段探测代码搞定这些参数:
import csv import chardet def sniff_csv_format(path: str, sample_bytes: int = 65536): """返回 csv 的编码、分隔符与是否有表头""" with open(path, 'rb') as f: raw = f.read(sample_bytes) encoding = chardet.detect(raw)['encoding'] or 'utf-8' text = raw.decode(encoding, errors='replace') try: dialect = csv.Sniffer().sniff(text[:4096], delimiters=',;\t') delimiter = dialect.delimiter has_header = csv.Sniffer().has_header(text[:4096]) except csv.Error: # 采样太规整时 Sniffer 可能报错,回退到逗号分隔 delimiter, has_header = ',', True return {'encoding': encoding, 'delimiter': delimiter, 'has_header': has_header} info = sniff_csv_format('data/raw_train.csv') print(info)errors='replace'是关键:采样文本如果被错误解码,至少不会抛异常中断,替换字符占位后仍然能从统计规律上给出编码判断。delimiters=',;\t'限定了候选分隔符,避免 Sniffer 把空格也算进去。对千万行级别的 csv 来说,读一个文件头通常足够判断,不必把整份数据加载进内存。
拿到探测结果后再真正导入:
import pandas as pd df = pd.read_csv( 'data/raw_train.csv', encoding=info['encoding'], # 探测出来的实际编码 sep=info['delimiter'], # 探测出来的分隔符 ) print(df.head()) print(df.dtypes)如果第一列列名带着\ufeff,那就是 BOM 没吃干净,把 encoding 改成'utf-8-sig'即可,pandas 会自动兼容。这个坑很小,但初学时很容易对着列名\ufeffuser_id找半天原因。
另外提醒一句,不少图像项目的 csv 里存的是image_path,label这样的文件列表,再交给后续 CNN 读取图片做训练。这类 csv 没有复杂统计列,但路径列经常因为 Windows 与 Linux 路径分隔符不一致读不出来,同样值得在读入时统一规范一次路径格式。
3.2 dtype 压缩与分块读取:大 CSV 不再吃光内存
csv 文件上到 1GB 就必然要面对内存问题。pandas 默认按内容猜类型,int64、float64 遍地走,而实际上用户 id 可能只有几十万种,状态列只有 0/1。把整数压缩成 int32 或 int8,内存占用直接砍掉一半以上。我一般先把 dtype 规划好再读入,不在训练时才去补救:
dtype_spec = { 'user_id': 'int32', 'is_active': 'int8', 'age': 'int8', 'price': 'float32', 'category': 'category', } df = pd.read_csv( 'data/raw_train.csv', dtype=dtype_spec, low_memory=False, # 让 pandas 一次性推断各列 dtype,而不是按块推断 ) print(df.memory_usage(deep=True) / 1024**2)low_memory=False在 C 引擎下会先采集全数据的类型推断表,再统一建 DataFrame,代价是第一次读取稍慢,但避免了按块推断时 int64 与 float64 混用的类型抖动。category非常适合字符串列有大量重复值的情形,比如城市名、渠道来源,pandas 内部做哈希映射,字符串内存占用大幅下降,对后续get_dummies和factorize也友好。
如果 csv 实在太大,比如超过可用内存的一半,就分块读入,把清洗动作写成一个函数逐块执行:
reader = pd.read_csv('data/huge.csv', dtype=dtype_spec, chunksize=500_000) clean_chunks = [] for chunk in reader: chunk = chunk.dropna(subset=['target']) chunk['price'] = chunk['price'].clip(0, 999) clean_chunks.append(chunk) df = pd.concat(clean_chunks, ignore_index=True)分块执行时要注意:dropna这类行过滤会让每个分块的统计口径不一致。如果后面要按 8:1:1 切数据集,最好先把划分键的随机种子和分块策略固定,再对每个分块做同样的过滤与类型转换。分块不意味着样本分布可以被忽略,尤其 csv 按时间或终端分组时,块间标签比例可能差得很远,这也是后续训练时验证集漂移的隐患之一。
如果你手里是多份 csv 需要合并,比如按月导出的一组训练数据,pandas 的 concat 是常见做法,但合并前要确保列名和列顺序一致。我会在 concat 之前先跑一个列名断言,防止某个月份的文件多了一列或少了一列,导致整个合并结果错位。
4. 把 CSV 改造成深度学习能用的数据集:清洗、编码和划分
4.1 缺失值、重复样本和异常值:清洗规则要写在脚本里
csv 数据直接喂给模型之前,必须过一遍清洗。这里的基准原则:清洗规则必须作为脚本保留,而不是在 Excel 里手工改。深度学习项目要复现,人工改一版就复现不了;而且手工误删一行,训练结果差得毫无头绪,事后完全追查不到。最常用的一组规则包括删除完全重复的样本、填充缺失值、对数值型异常做截断:
import numpy as np def clean_dataframe(df): # 1. 完全重复样本只保留一条 df = df.drop_duplicates(subset=['sample_id'], keep='first') # 2. 缺失值处理:数值列用中位数,类别列用特殊占位符 num_cols = df.select_dtypes(include=[np.number]).columns.tolist() cat_cols = df.select_dtypes(exclude=[np.number]).columns.tolist() df[num_cols] = df[num_cols].fillna(df[num_cols].median()) df[cat_cols] = df[cat_cols].fillna('UNKNOWN') # 3. 数值异常截断:低于 1% 分位和高于 99% 分位的值拉回边界 for col in num_cols: lo = df[col].quantile(0.01) hi = df[col].quantile(0.99) if np.isfinite(lo) and np.isfinite(hi): df[col] = df[col].clip(lo, hi) return df用中位数而不是均值填充,是避免个别超大值把填充值带偏。对类别列填UNKNOWN而不是直接删整行,因为深度学习模型可以对未知状态做出响应,但你不能平白少一条样本。分位截断用的是 0.01/0.99,属于经验默认值;对价格这种长尾分布,我通常先做np.log1p对数变换再剪分位,分布会更接近正态。clip只改变极值,不动大多数样本分布,所以是最安全的一类清洗操作。
还有一个容易被忽略的点:清洗顺序要固定。先 drop_duplicates 再 fillna 与先 fillna 再 drop 的结果不一样,去重后索引会重新排列还是保留原索引也影响后续划分。我的习惯是清洗函数一次性跑完,输出结果另存为一个 intermediate.csv,之后所有训练验证都用这份中间文件,不反复改原始数据。
4.2 标签编码与数据集划分:别让验证集泄露信息
表格分类任务的标签常常是字符串,比如风险等级、流失与否。模型只认数字,所以要做一次 label 编码:
classes = sorted(df['label'].unique()) label2id = {c: i for i, c in enumerate(classes)} df['label_id'] = df['label'].map(label2id) import json with open('label2id.json', 'w', encoding='utf-8') as f: json.dump(label2id, f, ensure_ascii=False, indent=2)这里的坑是:classes = sorted(df['label'].unique())必须在合并后的完整数据集上执行一次,然后保存映射文件。之后训练集、验证集、线上推理全部用同一个 label2id 做 map。如果验证集单独读一份 csv、单独 sorted,类别顺序变了,模型的输出维度可能没变,但语义逐一对调,训练过程不报错,只有效果莫名变差。这类问题找起来比模型本身还玄学,所以我把映射字典落盘成为项目的一部分。
数据集划分这一步我通常用 sklearn 的 train_test_split,注意两个容易被忽略的参数:
from sklearn.model_selection import train_test_split train_df, val_df = train_test_split( df, test_size=0.2, random_state=42, stratify=df['label_id'], # 按标签比例分层抽样 )stratify是不平衡数据集的后悔药。很多 csv 分类数据集正负样本比是 100:1,不按标签分层,随机划分可能让验证集里根本出现不了某类样本,训练分数虚高或莫名偏低。我基本把它当默认参数用。另外如果 csv 自带时间列,比如行为日志、订单记录,先按时间排序再划分,否则未来数据混进了训练集,线上效果一定会明显低于测试结果。时序数据也不要用随机切分,直接按时间点截断更可靠。
5. 避坑:从 CSV 到深度学习模型最常见的 5 个翻车现场
5.1 现象:loss 不下降,先查数据有没有乱序
训练开始后 loss 一直不降,或者降到一半卡住。大多数人第一反应是调学习率、换 optimizer,其实应该先看数据顺序。很多 csv 导出工具按标签把同类样本排在一起,train_test_split 之后如果不打乱,DataLoader 一个 batch 里全是同一个类别。模型在这个 batch 上学到的梯度方向单一,下一步又把其他类别全部推翻,loss 来回震荡。解决方法是 DataLoader 开shuffle=True:
from torch.utils.data import DataLoader loader = DataLoader(dataset, batch_size=64, shuffle=True)如果你不用 DataLoader,而是手动切 batch,那就在切分之前先把 DataFrame 打乱一遍:df.sample(frac=1, random_state=42)。注意打乱要在划分之后、喂数据之前执行,顺序别反过来。
5.2 现象:数据增强后标签错位
图像类 csv 项目通常把image_path,label存进 csv,训练时读图再增强。最容易翻车的位置是:你用 numpy 读图后,把 image 和 label 各自传给 transform,增强只处理 image,label 还是旧数组的索引,或者增强函数内部复制的只是引用。结果图像翻转或裁剪了,标签留在原位。解决方法是让增强逻辑对图像和标签同步处理,样本以字典形式传入 transform,增强函数只改图像不改标签数组。还要注意多进程 DataLoader 下,不要在__getitem__外维护一个与样本同步的可变全局列表,多进程复制之后标签会在某个 worker 里悄悄错位,这类问题最难排查。
5.3 现象:csv 中字符串列导致 embedding 维度对不上
类别列用pd.get_dummies做 one-hot,到了验证阶段读新 csv,发现里面出现训练时没见过的类别值,get_dummies 自动生成新列,模型输入维度比训练多了一列,直接报维度不匹配。原因是你没有固定类别字典。解决方法是第一,训练阶段把类别列转成 pandas 的categorydtype 并固定 categories;第二,像上一章那样保存 label2id 映射,推理时遇到未见类别映射到UNKNOWN。对取值很多的稀疏类别,我更建议用factorize加 embedding 表,不让 one-hot 维度爆炸。在训练脚本里加一段防御式检查也可以:
train_codes = set(df['city'].unique()) val_codes = set(val_df['city'].unique()) novel = val_codes - train_codes if novel: print(f"验证集出现训练集没有的类别: {novel}")5.4 现象:训练时 OOM,但 csv 明明不大
csv 只有 800MB,训练时显存或内存却炸了。正常情况下 800MB 的 csv 用 dtype 压缩后实际占用会低于这个数,OOM 多半是管道里某一步把数据复制成高精度数组。比如 pandas 的 float64 转 numpy 时没指定 dtype,或者读图时把每张图复制成 float32 的 4 通道。解决方法是读取 csv 时就按第三章的 dtype 方案压缩,转 numpy 时显式astype('float32'),DataLoader 的num_workers从 4 降到 2 减少内存拷贝。真遇到降不下去的情况,先定位哪列最占内存:
print(df.memory_usage(deep=True).sort_values(ascending=False))最占内存的往往就是没压缩的字符串列,把它换成 category dtype 立竿见影。
5.5 现象:zip 解压后文件 CRC 报错
用 Python 解压到一半抛Bad CRC-32,或者 csv 文件解压出来只有几 KB,而原始信息显示几百 MB。原因通常是网络传输断点续传损坏、打包工具差异或磁盘空间不足。先别急着重新下载,检查解压目标磁盘空间,df -h看一眼;再用 zipfile 自带 CRC 校验定位损坏的 entry。上面 2.1 的 safe_extract 在 extractall 时会自动校验 CRC,报错会直接指出哪个文件坏了。确认是单文件损坏时,可以向来源重新请求那一份 csv,没必要整个 zip 重下。
6. 进阶:用getitem实现 CSV 高效读取,并先跑通一个batch
6.1 自定义 Dataset 的最小实现
前面的 pandas 读取输出完整 DataFrame,对中小型 csv 没问题,但训练任务需要按需读取单行或小批量时,用自定义 Dataset 更合适:
from torch.utils.data import Dataset import torch class CsvDataset(Dataset): def __init__(self, df, feature_cols, label_col): self.features = torch.tensor( df[feature_cols].astype('float32').values, dtype=torch.float32, ) self.labels = torch.tensor( df[label_col].astype('int64').values, dtype=torch.long, ) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.features[idx], self.labels[idx]这里先把整列转成 torch tensor,__getitem__的按行切片开销很小。astype('float32')是必要一步,直接.values会得到 float64,训练时网络参数是 float32,计算图会在 loss 函数里频繁隐式转换,不报错但速度和显存都吃亏。feature_cols的列顺序必须先固定,模型训练后权重与新数据列顺序对不上时,错误会非常隐蔽。
6.2 模型前先验证数据管道
不管 csv 是十兆还是十吉,装进 Dataset 之后,训练之前至少做一次“喂一个 batch”的自检:
from torch.utils.data import DataLoader def sanity_check_dataset(dataset, batch_size=16): loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) x, y = next(iter(loader)) assert x.shape[0] == batch_size assert x.dtype == torch.float32 assert y.dtype == torch.long assert torch.isfinite(x).all(), "特征里出现 NaN 或 Inf" print(f"自检通过:x shape={tuple(x.shape)}, y range={y.min().item()}~{y.max().item()}") sanity_check_dataset(CsvDataset(train_df, FEATURE_COLS, 'label_id'))只取一个 batch 的代价几乎为零,却能触发 csv 读取、类型转换与 DataLoader 采样的完整链路。我习惯把这段检查写进训练脚本入口,每次跑之前先过一遍,失败就不往下跑。没有这个习惯之前,我在 CSV 数据管道上翻车的次数不少,多数是列顺序或 dtype 这种小问题,早点暴露能省下大把排查时间。希望帮到你。
本文还有配套的精品资源,点击获取