AI数据工程学习：用Python搭建可验证的数据清洗流水线
AI数据工程学习的重点不是先跑模型，而是让原始数据变成格式统一、缺失可控、可重复处理的数据集。
本文在Linux服务器上用Python和pandas搭建一条最小可用的数据清洗流水线，适合刚接触AI数据工程、需要把CSV或日志数据整理成训练输入的新手。
按步骤执行后，你能得到一份清洗报告和一份干净的数据文件，并能根据报告定位问题。
先确认运行环境和数据边界
在开始写脚本前，需要确认三件事：Python版本、依赖库和数据文件位置。
以下命令在Ubuntu 22.04或Debian 12上验证过，其他发行版路径可能略有差异。
python3 --version
pip3 --version
mkdir -p ~/ai-data-lab/raw ~/ai-data-lab/clean预期结果是Python 3.8以上、pip可用，raw目录用来放原始数据，clean目录存放清洗后的结果。
接着安装pandas和openpyxl，后者用于读取Excel文件；
如果只处理CSV可以不装。
pip3 install pandas openpyxl如果提示externally-managed-environment，说明系统Python受保护，建议用虚拟环境：
python3 -m venv ~/ai-data-lab/venv
source ~/ai-data-lab/venv/bin/activate
pip install pandas openpyxl准备一份示例数据，保存为~/ai-data-lab/raw/sample.csv，包含id,name,age,city,score五列，故意留几行缺失值和重复id。
这一步不需要真实业务数据，重点是让后续脚本有可检查的对象。
编写清洗脚本并逐项检查
新建~/ai-data-lab/clean_data.py，按读取、去重、缺失处理、类型转换、导出五个阶段组织。
下面代码可直接复制，注释说明每一步的预期结果。
import pandas as pd
RAW = "/home/你的用户名/ai-data-lab/raw/sample.csv"
CLEAN = "/home/你的用户名/ai-data-lab/clean/clean.csv"
REPORT = "/home/你的用户名/ai-data-lab/clean/report.txt"
df = pd.read_csv(RAW)
print("原始行数:", len(df))
# 1. 去重：以id为准保留第一条
df = df.drop_duplicates(subset="id", keep="first")
# 2. 缺失处理：age用中位数填充，city用未知填充
df["age"] = df["age"].fillna(df["age"].median())
df["city"] = df["city"].fillna("未知")
# 3. 类型转换：age转整数，score转浮点
df["age"] = df["age"].astype(int)
df["score"] = pd.to_numeric(df["score"], errors="coerce")
# 4. 导出并生成报告
df.to_csv(CLEAN, index=False)
with open(REPORT, "w") as f:
f.write(f"清洗后行数: {len(df)}\n")
f.write(f"缺失值统计:\n{df.isnull().sum().to_string()}\n")
print("清洗完成，输出:", CLEAN)运行脚本：
cd ~/ai-data-lab
python3 clean_data.py预期终端打印原始行数和清洗完成提示，clean目录下出现clean.csv和report.txt。
如果报FileNotFoundError，检查RAW路径中的用户名是否替换正确；
如果报KeyError: 'id'，说明CSV表头与代码列名不一致，用head -1 raw/sample.csv查看实际表头。
常见报错和避坑清单
数据清洗脚本跑不通，多数不是逻辑问题，而是路径、编码和类型边界。
下面几条是新手最容易踩的坑。
- 路径写死绝对路径：脚本里用了
/home/你的用户名/，复制后必须替换成实际用户目录，或者改用os.path.expanduser("~/ai-data-lab/raw/sample.csv")。 - 中文编码报错：CSV含中文时，读取加
encoding="utf-8"，导出同样指定；如果源文件是GBK，先转存为UTF-8再处理。 - 缺失值填充后类型仍报错：
fillna之后如果整列都是空值，中位数也是NaN，astype(int)会失败。先用df["age"].notna().sum()确认有效值数量。 - 去重把有效数据删掉：
drop_duplicates前先明确业务主键，不要直接对整行去重，否则相同内容但不同id的记录会被误删。 - 不要直接覆盖原始文件：清洗结果统一写到
clean目录，保留raw原始数据，方便回滚和对比。
如果脚本中途失败，先看报错最后一行定位文件和行号，再回到对应步骤检查输入数据，不要盲目重跑整个流程。
验证清洗结果是否可用
脚本跑完不等于数据可用，需要做三项检查。
第一，对比行数变化：wc -l clean/clean.csv与原始文件行数比较，差值应等于去重和无效行删除的数量。
第二，查看报告中的缺失值统计，report.txt里各列缺失数应为0或符合预期。
第三，抽样检查关键列：
head -5 clean/clean.csv
python3 -c "import pandas as pd; df=pd.read_csv('clean/clean.csv'); print(df.dtypes); print(df.describe())"预期age为整数类型、score为浮点类型，且describe()输出的最小值和最大值在合理范围内。
如果score出现大量NaN，说明原始数据中有非数字字符被coerce转成了空值，需要回到原始文件确认这些行是否应该保留。
验证通过后，这份clean.csv就可以作为后续特征工程或模型训练的输入。
AI数据工程学习的下一步通常是划分训练集和验证集，但前提是当前清洗结果稳定、可重复。
建议把脚本和报告一起存档，每次换数据只改输入路径，不改清洗逻辑，这样才能保证不同批次数据经过同一套处理。
如果后续数据量增大，可以把pandas换成Dask或PySpark，但清洗规则和验证项应保持一致。
先跑通单机小数据流程，再考虑分布式，是更稳妥的学习路径。
