3. 数据加载与预处理

数据加载与预处理，说白了就是推理管线的「第一道关卡」。我见过太多团队，模型选得挺好，推理框架也搭得漂亮，结果数据加载环节成了瓶颈——GPU 空转，CPU 忙死。嗯，这章我们就来聊聊怎么把这关打通。

3.1 批量数据读取：CSV、Parquet、JSON

推理场景下，数据源五花八门。我个人习惯先把数据格式理清楚，再选读取工具。下面是我常用的三种格式和对应的读取方式。

3.1.1 CSV 读取

CSV 是最常见的格式，但坑也不少。我曾经遇到一个项目，CSV 文件里某列混入了换行符，导致 pandas 读取直接崩掉。后来我养成了一个习惯：读取时指定 dtype 和 na_values，避免类型推断带来的意外。

import pandas as pd

# 推荐：指定列类型，避免自动推断
dtype_spec = {
    'user_id': 'int32',
    'age': 'float32',
    'score': 'float32'
}
df = pd.read_csv('data.csv', dtype=dtype_spec, na_values=['', 'NA', 'null'])
print(f"加载完成，共 {len(df)} 行")
小技巧：如果 CSV 文件很大，可以用 chunksize 参数分块读取，避免内存爆炸。我一般设成 10000 行一块。

3.1.2 Parquet 读取

Parquet 是我在推理项目中的首选格式。为什么？因为它列式存储 + 压缩，读取速度比 CSV 快 3-5 倍。你想想看，推理时我们往往只需要部分列，Parquet 能直接跳过无关列，省带宽省内存。

import pyarrow.parquet as pq

# 只读取需要的列，减少 I/O
table = pq.read_table('data.parquet', columns=['feature_1', 'feature_2', 'label'])
df = table.to_pandas()
print(f"Parquet 加载完成，列数：{len(df.columns)}")
注意：Parquet 文件如果由 Spark 生成，分区数可能很多。读取时建议用 use_threads=True 开启多线程，否则单线程读会慢得让你怀疑人生。

3.1.3 JSON 读取

JSON 在日志和 API 场景很常见。但 JSON 嵌套结构多，直接读成 DataFrame 容易出问题。我一般先解析成 Python 字典，再展平。

import json
import pandas as pd

with open('data.json', 'r') as f:
    records = [json.loads(line) for line in f]  # 每行一个 JSON 对象

# 展平嵌套字段
def flatten(record, parent_key='', sep='_'):
    items = []
    for k, v in record.items():
        new_key = f"{parent_key}{sep}{k}" if parent_key else k
        if isinstance(v, dict):
            items.extend(flatten(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

flat_records = [flatten(r) for r in records]
df = pd.DataFrame(flat_records)

3.2 数据清洗与标准化

数据清洗，说白了就是「去脏」。我总结了一个三步走策略：缺失值处理 → 异常值过滤 → 标准化。

3.2.1 缺失值处理

推理时遇到缺失值，不能像训练时那样随便填充。我曾经犯过一个错：用均值填充缺失值，结果推理结果全偏了。后来我改用前向填充 + 标记位的方式。

# 前向填充，同时记录哪些位置被填充过
df['feature_1'] = df['feature_1'].fillna(method='ffill')
df['feature_1_was_missing'] = df['feature_1'].isna().astype(int)
核心原则：推理时的缺失值处理策略，必须和训练时完全一致。否则模型会「看不懂」输入。

3.2.2 异常值过滤

异常值在推理时很要命。我习惯用 IQR（四分位距）方法做初步过滤，但注意不要一刀切——有些业务场景下「异常值」反而是关键信号。

Q1 = df['score'].quantile(0.25)
Q3 = df['score'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常值，而不是直接删除
df['is_outlier'] = (df['score'] < lower_bound) | (df['score'] > upper_bound)

3.2.3 标准化

标准化是推理前的最后一步。我建议用 Z-score 标准化，但均值和标准差必须从训练集计算，推理时直接复用。

# 假设训练集统计量已保存
mean_val = 0.5
std_val = 0.2

df['score_normalized'] = (df['score'] - mean_val) / std_val

3.3 特征工程与向量化

特征工程，说白了就是把原始数据变成模型能「吃」的格式。我把它分成两类：数值特征和类别特征。

3.3.1 数值特征处理

数值特征除了标准化，还可以做分箱或对数变换。比如用户年龄，直接输入模型效果一般，但分箱后效果立竿见影。

import numpy as np

# 对数变换：处理长尾分布
df['amount_log'] = np.log1p(df['amount'])

# 分箱：将连续值离散化
bins = [0, 18, 30, 45, 60, 100]
labels = ['0-18', '19-30', '31-45', '46-60', '60+']
df['age_bin'] = pd.cut(df['age'], bins=bins, labels=labels)

3.3.2 类别特征处理

类别特征我推荐用 Label Encoding 或 Target Encoding。One-Hot 编码在类别数多的时候会爆炸，不推荐。

from sklearn.preprocessing import LabelEncoder

le = LabelEncoder()
df['city_encoded'] = le.fit_transform(df['city'])

# 注意：推理时要用同一个 LabelEncoder 对象
# 保存 le 到文件，推理时加载

3.3.3 向量化

文本特征需要向量化。我常用 TF-IDF 或 预训练词向量。注意：TF-IDF 的词汇表必须和训练时一致。

from sklearn.feature_extraction.text import TfidfVectorizer

# 假设训练时已拟合的 vectorizer
vectorizer = TfidfVectorizer(max_features=5000)
vectorizer.fit(train_texts)  # 训练时拟合

# 推理时直接 transform
X_text = vectorizer.transform(df['text_column'])

3.4 使用 DataLoader 进行批量加载

DataLoader 是批量推理的核心组件。我习惯自己封装一个，支持预取和多线程。

3.4.1 自定义 DataLoader

import torch
from torch.utils.data import Dataset, DataLoader

class InferenceDataset(Dataset):
    def __init__(self, df, feature_cols):
        self.data = df[feature_cols].values.astype('float32')
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, idx):
        return self.data[idx]

dataset = InferenceDataset(df, feature_cols=['feature_1', 'feature_2', 'feature_3'])
loader = DataLoader(
    dataset,
    batch_size=64,
    shuffle=False,       # 推理时不要 shuffle
    num_workers=4,       # 多进程加载
    pin_memory=True      # 加速 GPU 传输
)
经验之谈：num_workers 一般设为 CPU 核心数的一半。设太多反而会因为上下文切换而变慢。

3.4.2 批量推理循环

model.eval()
all_predictions = []

with torch.no_grad():
    for batch in loader:
        batch = batch.to('cuda')
        outputs = model(batch)
        all_predictions.append(outputs.cpu().numpy())

predictions = np.concatenate(all_predictions, axis=0)

3.5 本章知识体系

下面这张图概括了数据加载与预处理的完整流程。我把它画成流程图，方便你对照着搭管线。

数据加载与预处理流程 数据源 数据读取（CSV / Parquet / JSON） 数据清洗（缺失值 / 异常值） 特征工程（标准化 / 编码 / 向量化） DataLoader 批量加载

嗯，数据加载与预处理这块，说白了就是「磨刀不误砍柴工」。管线搭好了，后面推理才能跑得稳。我建议你从 CSV 或 Parquet 开始，先把清洗和标准化跑通，再逐步加上特征工程和 DataLoader。别一上来就想搞全套，容易翻车。

一句话总结：数据加载要快，清洗要准，特征要稳，批量要狠。
← 上一章 📖 返回目录 下一章 →