机器学习入门：任务调度 数据预处理

前段时间开始折腾机器学习，跟着教程走的时候一切都挺美好，数据集是现成的，load_iris() 一调，fit 完直接出准确率。等真拿自己的数据上手才发现，数据这玩意儿才是最磨人的。缺失值要填，异常值要剔，类别要编码，数值要归一化，而且每次改个参数就得从头手动再跑一遍，一天下来大半时间都耗在这上头了。

后来干脆把预处理写成脚本，再挂到定时任务里让它自己跑，才算是解脱了。这篇就记录一下整个过程，主要涉及两块：任务调度和数据预处理，都是实测可用的，直接抄作业也没问题。

先说任务调度

机器学习的任务有个特点：数据天天在变，模型就得跟着重新训练。总不能每天早上爬起来手动执行一遍脚本吧，所以需要任务调度，说白了就是定个时间，让机器自动帮你跑。

常见方案就两个，一个是 Linux 自带的 crontab，一个是 Python 的 APScheduler。

如果只是每天定时跑一次，crontab 最省事，crontab -e 打开后加一行：

0 2 * * * /usr/bin/python3 /home/xxx/pipeline.py >> /home/xxx/pipeline.log 2>&1

意思是每天凌晨两点跑一次 pipeline.py，输出追加到日志里。后面那段日志重定向千万别省，不然脚本半夜挂了你根本不知道，第二天看着模型没更新还以为是数据有问题，排查半天发现压根就没跑。

要是逻辑再复杂点，比如任务之间有依赖，跑完预处理还得接着训练，APScheduler 用起来更舒服：

from apscheduler.schedulers.blocking import BlockingScheduler

scheduler = BlockingScheduler()

scheduler.add_job(run_pipeline, "cron", hour=2)
scheduler.add_job(train_model, "cron", hour=3)

scheduler.start()

这样凌晨两点先做数据预处理，三点再接着训练模型，串行执行，不用操心上一个任务还没跑完下一个就抢着开始的问题。

数据预处理到底在处理啥

拿到的原始数据基本都不能直接喂给模型，好在这活儿用 pandas 干还算顺手，核心就下面几步。

第一步先看一眼数据长啥样：

import pandas as pd

df = pd.read_csv("raw_data.csv")
print(df.info())
print(df.isnull().sum())

isnull().sum() 这一步别跳过，很多坑看这一眼就能暴露。比如某个字段缺了三成，那就不是填个均值能糊弄过去的事了，得考虑直接把这一列丢掉。

缺失值我这里简单处理，数值列填中位数，类别列填众数：

num_cols = df.select_dtypes(include="number").columns
cat_cols = df.select_dtypes(include="object").columns

for col in num_cols:
    df[col] = df[col].fillna(df[col].median())

for col in cat_cols:
    df[col] = df[col].fillna(df[col].mode()[0])

类别特征得转成数字，不然模型不认，独热编码一行搞定：

df = pd.get_dummies(df, columns=cat_cols)

数值特征做个标准化。不做的话，有的列范围是 0 到 1，有的列动辄上万，模型会被大数值的列带着跑偏：

from sklearn.preprocessing import StandardScaler

scaler = StandardScaler()
df[num_cols] = scaler.fit_transform(df[num_cols])

这里有个小坑提醒一下：scaler 训练完一定要存下来，比如 joblib.dump(scaler, "scaler.pkl")。因为预测的时候得用同一个 scaler 做变换，存都没存的话，线上数据和训练数据就对不上了，模型效果稀里糊涂变差，没准儿你排查半天都找不着原因。

最后把干净的数据存出去：

df.to_csv("clean_data.csv", index=False)

把两块拼起来

整个 pipeline.py 大概长这样：

import pandas as pd
import joblib
from sklearn.preprocessing import StandardScaler

def run_pipeline():
    df = pd.read_csv("raw_data.csv")

    num_cols = df.select_dtypes(include="number").columns
    cat_cols = df.select_dtypes(include="object").columns

    for col in num_cols:
        df[col] = df[col].fillna(df[col].median())
    for col in cat_cols:
        df[col] = df[col].fillna(df[col].mode()[0])

    df = pd.get_dummies(df, columns=cat_cols)

    scaler = StandardScaler()
    df[num_cols] = scaler.fit_transform(df[num_cols])
    joblib.dump(scaler, "scaler.pkl")  # <--- 预测时要加载同一个 scaler

    df.to_csv("clean_data.csv", index=False)

if __name__ == "__main__":
    run_pipeline()

不出问题的话就没有问题了，第二天早上起来瞄一眼日志，处理好的 clean_data.csv 已经躺在目录里了，直接拿去训练就行。自从挂上定时任务过后，我是再也没手动跑过预处理，这时间省下来干啥不好。