4. 数据集准备：使用公开数据集模拟生产数据流

在特征漂移监控实战中，数据集准备是构建可靠监控管线的第一步。本章选用 UCI Adult Income（又称“人口普查收入”）数据集，该数据集包含 48,842 条记录，混合了数值型与类别型特征，天然适合模拟生产环境中的结构化数据流。我们将完成以下三项核心任务：

• 数据获取与清洗：下载原始 CSV，处理缺失值与异常编码。
• 训练集与实时批次划分：按时间或随机顺序切分，模拟“历史训练 + 在线推理”场景。
• 数据流模拟器构建：将测试批次包装为逐批到达的生成器，用于后续漂移检测。

4.1 数据集概览与字段说明

下表列出了 Adult Income 数据集中用于漂移监控的关键字段及其类型。注意，我们将 income 作为目标变量，其余字段作为特征。

4.2 数据流模拟架构

为了贴近生产环境，我们不能一次性加载全部测试数据，而应将其包装成“实时批次”。下图展示了从原始数据集到模拟数据流的完整架构：

4.3 代码实现：从 CSV 到数据流

以下 Python 代码实现了完整的准备流程。我们使用 pandas 进行数据处理，并构建一个生成器函数 data_stream() 来模拟实时到达的批次。

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder

# 1. 加载原始数据（假设已下载 adult.data）
columns = ['age','workclass','fnlwgt','education','education-num',
           'marital-status','occupation','relationship','race','sex',
           'capital-gain','capital-loss','hours-per-week','native-country','income']
df = pd.read_csv('adult.data', names=columns, sep=',\s*', engine='python')

# 2. 数据清洗：去除缺失值（用 '?' 表示）
df.replace('?', np.nan, inplace=True)
df.dropna(inplace=True)

# 3. 选择用于漂移监控的特征子集
feature_cols = ['age','workclass','education','marital-status',
                'occupation','capital-gain','hours-per-week']
target_col = 'income'

# 4. 对类别特征进行标签编码
le_dict = {}
for col in feature_cols:
    if df[col].dtype == 'object':
        le = LabelEncoder()
        df[col] = le.fit_transform(df[col].astype(str))
        le_dict[col] = le

# 5. 划分训练集与测试集（70% / 30%）
X = df[feature_cols]
y = (df[target_col] == '>50K').astype(int)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, random_state=42, stratify=y
)

print(f"训练集大小: {X_train.shape[0]}")
print(f"测试集大小: {X_test.shape[0]}")

# 6. 构建数据流模拟器
def data_stream(X, y, batch_size=500, shuffle=True):
    """
    生成器函数：每次 yield 一个批次的特征和标签。
    """
    data = pd.concat([X, y], axis=1).values
    if shuffle:
        np.random.shuffle(data)
    for i in range(0, len(data), batch_size):
        batch = data[i:i+batch_size]
        yield batch[:, :-1], batch[:, -1]

# 7. 使用示例：模拟前 3 个批次
stream = data_stream(X_test, y_test, batch_size=500)
for batch_idx, (X_batch, y_batch) in enumerate(stream):
    if batch_idx >= 3:
        break
    print(f"批次 {batch_idx+1}: X_shape={X_batch.shape}, y_mean={y_batch.mean():.3f}")

4.4 批次划分策略对比

在实际生产中，数据到达的顺序可能影响漂移检测的敏感性。下表对比了两种常见的批次划分策略：

4.5 验证数据流完整性

在进入漂移检测之前，务必验证数据流是否按预期工作。以下检查清单可以帮助你快速排查问题：

• 批次大小一致性：除最后一个批次外，所有批次应具有相同的行数。
• 特征顺序不变：每个批次的特征列顺序必须与训练集完全一致。
• 类别编码对齐：测试批次中的类别值必须在训练集的编码范围内，避免出现未见过的标签。
• 目标变量分布：打印每个批次中正例（>50K）的比例，确保没有极端失衡。

# 快速验证：检查批次大小与类别编码
stream = data_stream(X_test, y_test, batch_size=500)
batch_sizes = []
for X_batch, y_batch in stream:
    batch_sizes.append(len(X_batch))
    # 检查类别特征是否在训练集编码范围内
    for col_idx in [1, 2, 3, 4]:  # workclass, education, marital-status, occupation
        unique_vals = set(X_batch[:, col_idx])
        train_vals = set(X_train.iloc[:, col_idx].unique())
        if not unique_vals.issubset(train_vals):
            print(f"警告：批次中发现训练集未出现的类别值！")

print(f"批次大小列表（前10个）: {batch_sizes[:10]}")
print(f"所有批次大小一致: {len(set(batch_sizes)) == 1 or batch_sizes[-1] <= 500}")