完整 ML 工作流：从数据到部署

摘要：前面七篇文章学完了经典机器学习的核心算法。但掌握算法只是第一步——在真实项目中，80% 的时间花在算法以外的事情上：理解业务、清洗数据、特征工程、模型评估、调参优化、部署上线。这篇文章把整个 ML 工作流串起来，用一个端到端的实战项目演示从原始数据到可部署模型的全流程。

一、ML 工作流全景

一个完整的机器学习项目通常包含以下阶段：

┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ 业务理解 │→│ 数据获取 │→│ 数据清洗 │→│ 特征工程 │→│ 模型训练 │→│ 模型评估 │
│ & 问题定义 │ │ & 探索分析 │ │ & 预处理 │ │ & 特征选择 │ │ & 调参优化 │ │ & 部署 │
└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
↑ ↑ ↑ ↑
最关键的一步 最耗时的一步 最影响效果 持续迭代
(错了全白做) (60% 的时间) (特征 > 模型)

各阶段时间占比（真实项目经验）

数据清洗 & 特征工程： 60% ← 最被低估的部分
模型训练 & 调参： 20% ← 算法知识集中在这
业务理解 & 问题定义： 10% ← 决定项目成败
部署 & 监控： 10% ← 容易被忽视

二、阶段 1：业务理解与问题定义

错误的开始

❌ "我要用 XGBoost 做分类"
→ 从算法出发，不是从问题出发

❌ "我的数据有这么多列，肯定能做出好模型"
→ 数据量 ≠ 数据质量

正确的开始

在写任何代码之前，先回答四个问题：

问题类型决定算法

将业务问题映射到 ML 问题类型：

# 业务问题 → ML 问题 → 算法选择

"预测下个月销售额" → 回归 → 线性回归 / 随机森林 / XGBoost
"客户会流失吗？" → 二分类 → 逻辑回归 / SVM / XGBoost
"客户属于哪类？" → 多分类 → Softmax / 随机森林 / XGBoost
"哪些客户群体？" → 聚类 → K-Means / DBSCAN
"产品 A 经常和什么一起买？" → 关联规则 → Apriori

三、阶段 2：数据获取与探索分析（EDA）

数据加载与初步探索

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline, make_pipeline
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, ConfusionMatrixDisplay, roc_auc_score
import warnings
warnings.filterwarnings('ignore')

# 加载数据
df = pd.read_csv('customer_churn.csv')
print(f"数据集大小: {df.shape}")
print(f"列名: {df.columns.tolist()}")

探索性数据分析（EDA）

# ===== 1. 目标变量分布 =====
df['churn'].value_counts().plot(kind='bar')
plt.title('目标变量分布（客户是否流失）')
plt.show()

# ===== 2. 缺失值检查 =====
missing = df.isnull().sum()
missing = missing[missing > 0].sort_values(ascending=False)
if len(missing) > 0:
print("缺失值情况：")
print(missing)

# ===== 3. 数值型特征分布 =====
df.hist(figsize=(15, 10), bins=30)
plt.tight_layout()
plt.show()

# ===== 4. 相关性矩阵 =====
plt.figure(figsize=(12, 10))
numeric_cols = df.select_dtypes(include=[np.number]).columns
sns.heatmap(df[numeric_cols].corr(), annot=True, fmt='.2f', cmap='RdBu')
plt.title('特征相关性矩阵')
plt.show()

# ===== 5. 特征与目标的关系 =====
fig, axes = plt.subplots(2, 3, figsize=(15, 10))
features_to_plot = ['tenure', 'monthly_charges', 'total_charges',
'contract_type', 'payment_method', 'internet_service']
for ax, feature in zip(axes.ravel(), features_to_plot):
if df[feature].dtype == 'object':
df.groupby(feature)['churn'].mean().sort_values().plot(kind='bar', ax=ax)
else:
df.boxplot(column=feature, by='churn', ax=ax)
ax.set_title(f'{feature} vs Churn')
ax.set_ylabel('流失率')
plt.tight_layout()
plt.show()

EDA 要点总结

# EDA 完成后，总结关键发现：
print("=" * 50)
print("EDA 关键发现：")
print("=" * 50)
print(f"1. 流失率: {df['churn'].mean():.1%}")
print(f"2. 缺失值列: {list(missing.index) if len(missing) > 0 else '无'}")
print(f"3. 强相关特征: {','.join(strong_features)}")
print(f"4. 类别特征数: {len(categorical_cols)}")
print(f"5. 数值特征数: {len(numeric_cols)}")

四、阶段 3：数据清洗与预处理

真实世界的数据是不完美的——缺失值、异常值、不一致的格式。

数据清洗流程

def clean_data(df):
"""数据清洗流水线"""
df = df.copy()

# 1. 删除完全无用的列
cols_to_drop = ['customer_id', 'phone_number'] # ID 类特征
df = df.drop(columns=[c for c in cols_to_drop if c in df.columns])

# 2. 处理特殊值
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

# 3. 处理异常值（以 tenure 为例）
Q1 = df['tenure'].quantile(0.25)
Q3 = df['tenure'].quantile(0.75)
IQR = Q3 – Q1
lower, upper = Q1 – 3*IQR, Q3 + 3*IQR
df = df[(df['tenure'] >= lower) & (df['tenure'] <= upper)]

return df

df_clean = clean_data(df)
print(f"清洗前: {len(df)}, 清洗后: {len(df_clean)}")

特征与目标分离 + 划分数据集

# 分离特征和目标
X = df_clean.drop('churn', axis=1)
y = df_clean['churn']

# 划分：训练集 60% + 验证集 20% + 测试集 20%
X_train, X_temp, y_train, y_temp = train_test_split(
X, y, test_size=0.4, random_state=42, stratify=y, shuffle=True
)
X_val, X_test, y_val, y_test = train_test_split(
X_temp, y_temp, test_size=0.5, random_state=42, stratify=y_temp
)

print(f"训练集: {len(X_train)}")
print(f"验证集: {len(X_val)}")
print(f"测试集: {len(X_test)}")

五、阶段 4：特征工程

特征工程是"最被低估但影响最大"的环节。好的特征比好模型更重要。

特征类型识别

# 自动识别特征类型
numeric_features = X_train.select_dtypes(include=[np.number]).columns.tolist()
categorical_features = X_train.select_dtypes(include=['object']).columns.tolist()

print(f"数值特征 ({len(numeric_features)}): {numeric_features}")
print(f"类别特征 ({len(categorical_features)}): {categorical_features}")

构建预处理流水线

# 数值特征流水线
numeric_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='median')), # 中位数填充缺失值
('scaler', StandardScaler()), # 标准化
])

# 类别特征流水线
categorical_transformer = Pipeline(steps=[
('imputer', SimpleImputer(strategy='most_frequent')), # 众数填充
('onehot', OneHotEncoder(handle_unknown='ignore', # 独热编码
sparse_output=False)),
])

# 合并为 ColumnTransformer
preprocessor = ColumnTransformer(
transformers=[
('num', numeric_transformer, numeric_features),
('cat', categorical_transformer, categorical_features),
]
)

# 查看预处理后的维度
X_train_processed = preprocessor.fit_transform(X_train)
print(f"预处理前特征数: {X_train.shape[1]}")
print(f"预处理后特征数: {X_train_processed.shape[1]}")

特征工程进阶技巧

# 1. 创建交叉特征（交互特征）
df['avg_monthly_charge'] = df['total_charges'] / (df['tenure'] + 1)

# 2. 分箱（将连续特征离散化）
df['tenure_group'] = pd.cut(df['tenure'],
bins=[0, 12, 24, 48, 72, 100],
labels=['<1年', '1-2年', '2-4年', '4-6年', '>6年'])

# 3. 对数变换（处理长尾分布）
df['log_total_charges'] = np.log1p(df['total_charges'])

# 4. 聚合特征（如果有多条记录/用户）
# user_avg = df.groupby('user_id')['amount'].mean().reset_index()

六、阶段 5：模型训练与调参

建立 Baseline

# 用默认参数的逻辑回归做 baseline
from sklearn.linear_model import LogisticRegression

baseline = Pipeline(steps=[
('preprocessor', preprocessor),
('classifier', LogisticRegression(max_iter=1000, random_state=42))
])

baseline.fit(X_train, y_train)
baseline_score = baseline.score(X_val, y_val)
print(f"Baseline（逻辑回归）验证集准确率: {baseline_score:.3f}")

多模型对比

from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
import xgboost as xgb

models = {
'逻辑回归': LogisticRegression(max_iter=1000, random_state=42),
'随机森林': RandomForestClassifier(n_estimators=200, random_state=42, n_jobs=-1),
'梯度提升': GradientBoostingClassifier(n_estimators=200, max_depth=3, random_state=42),
'XGBoost': xgb.XGBClassifier(n_estimators=200, max_depth=3, random_state=42, n_jobs=-1),
}

results = []
for name, model in models.items():
pipeline = Pipeline(steps=[
('preprocessor', preprocessor),
('classifier', model)
])

pipeline.fit(X_train, y_train,
classifier__eval_set=[(X_val, y_val)] if name == 'XGBoost' else None,
classifier__verbose=False)

train_acc = pipeline.score(X_train, y_train)
val_acc = pipeline.score(X_val, y_val)

results.append({'模型': name, '训练准确率': train_acc, '验证准确率': val_acc})

results_df = pd.DataFrame(results).round(3)
print(results_df.to_string(index=False))

网格搜索调参

# 以 XGBoost 为例
xgb_pipeline = Pipeline(steps=[
('preprocessor', preprocessor),
('classifier', xgb.XGBClassifier(random_state=42, n_jobs=-1))
])

param_grid = {
'classifier__n_estimators': [100, 200, 300],
'classifier__max_depth': [3, 5, 7],
'classifier__learning_rate': [0.01, 0.05, 0.1],
'classifier__subsample': [0.8, 1.0],
'classifier__colsample_bytree': [0.8, 1.0],
}

grid_search = GridSearchCV(
xgb_pipeline,
param_grid,
cv=3, # 3 折交叉验证
scoring='roc_auc', # 用 AUC 做优化目标（适合不平衡数据）
n_jobs=-1,
verbose=1
)

grid_search.fit(X_train, y_train)

print(f"\\n最佳参数: {grid_search.best_params_}")
print(f"最佳交叉验证 AUC: {grid_search.best_score_:.3f}")

七、阶段 6：模型评估

不仅仅看准确率

best_model = grid_search.best_estimator_

# 在测试集上做最终评估
y_pred = best_model.predict(X_test)
y_prob = best_model.predict_proba(X_test)[:, 1]

print("=" * 50)
print("最终模型评估（测试集）")
print("=" * 50)
print(f"AUC: {roc_auc_score(y_test, y_prob):.3f}")
print(f"\\n分类报告:")
print(classification_report(y_test, y_pred))

# 混淆矩阵
ConfusionMatrixDisplay.from_estimator(
best_model, X_test, y_test,
cmap='Blues',
display_labels=['未流失', '流失']
)
plt.title('测试集混淆矩阵')
plt.show()

特征重要性分析

# 提取特征重要性
if hasattr(best_model.named_steps['classifier'], 'feature_importances_'):
# 获取预处理后的特征名
cat_features_encoded = list(
best_model.named_steps['preprocessor']
.named_transformers_['cat']
.named_steps['onehot']
.get_feature_names_out(categorical_features)
)
all_features = numeric_features + cat_features_encoded

importance = pd.DataFrame({
'feature': all_features,
'importance': best_model.named_steps['classifier'].feature_importances_
}).sort_values('importance', ascending=False).head(15)

plt.figure(figsize=(10, 6))
plt.barh(importance['feature'], importance['importance'])
plt.xlabel('重要性')
plt.title('Top 15 特征重要性')
plt.gca().invert_yaxis()
plt.tight_layout()
plt.show()

模型评估清单

def evaluate_model_comprehensive(model, X_test, y_test):
"""全面的模型评估"""
y_pred = model.predict(X_test)
y_prob = model.predict_proba(X_test)[:, 1]

print("=" * 50)
print("模型评估报告")
print("=" * 50)

# 1. 分类指标
print(f"\\n准确率: {accuracy_score(y_test, y_pred):.3f}")
print(f"AUC: {roc_auc_score(y_test, y_prob):.3f}")
print(f"\\n分类报告:")
print(classification_report(y_test, y_pred))

# 2. 阈值优化
from sklearn.metrics import precision_recall_curve
precisions, recalls, thresholds = precision_recall_curve(y_test, y_prob)

f1_scores = 2 * precisions[:-1] * recalls[:-1] / (precisions[:-1] + recalls[:-1] + 1e-10)
best_threshold = thresholds[np.argmax(f1_scores)]
y_pred_opt = (y_prob >= best_threshold).astype(int)

print(f"最优阈值: {best_threshold:.3f}（F1最大化）")
print(f"优化后 F1: {f1_scores.max():.3f}")

return best_threshold

八、阶段 7：部署准备

保存模型

import joblib

# 保存完整流水线（包含预处理 + 模型）
joblib.dump(best_model, 'churn_model_pipeline.pkl')
print("模型已保存: churn_model_pipeline.pkl")

# 保存列名（用于 API 输入验证）
model_metadata = {
'numeric_features': numeric_features,
'categorical_features': categorical_features,
'threshold': best_threshold,
'model_type': type(best_model.named_steps['classifier']).__name__,
}
joblib.dump(model_metadata, 'model_metadata.pkl')

模型推理函数

def predict_churn(model, metadata, customer_data):
"""
生产环境推理函数

参数:
customer_data: dict 或 DataFrame，包含所有特征
返回:
prediction: 0（未流失）或 1（流失）
probability: 流失概率
"""
# 输入验证
required_cols = metadata['numeric_features'] + metadata['categorical_features']
missing_cols = [c for c in required_cols if c not in customer_data.columns]
if missing_cols:
raise ValueError(f"缺少特征: {missing_cols}")

# 预测
prob = model.predict_proba(customer_data)[:, 1][0]
pred = (prob >= metadata['threshold']).astype(int)

return {
'prediction': int(pred),
'probability': float(prob),
'risk_level': 'high' if prob > 0.7 else ('medium' if prob > 0.3 else 'low')
}

# 测试推理
sample = X_test.iloc[0:1]
result = predict_churn(best_model, model_metadata, sample)
print(f"推理结果: {result}")

简易 API 示例

from flask import Flask, request, jsonify
import joblib
import pandas as pd

app = Flask(__name__)

# 加载模型
model = joblib.load('churn_model_pipeline.pkl')
metadata = joblib.load('model_metadata.pkl')

@app.route('/predict', methods=['POST'])
def predict():
data = request.get_json()
df = pd.DataFrame([data])

try:
result = predict_churn(model, metadata, df)
return jsonify(result)
except Exception as e:
return jsonify({'error': str(e)}), 400

@app.route('/health', methods=['GET'])
def health():
return jsonify({'status': 'ok'})

# 启动: python app.py
# 请求: curl -X POST -H "Content-Type: application/json" -d '{"tenure": 12, "monthly_charges": 70}' http://localhost:5000/predict

九、ML 工作流清单

每当你开始一个新 ML 项目，按照这个清单逐项检查：

项目启动

• 业务目标是什么？成功标准是什么？
• 当前解决这个问题的方式是什么？
• ML 方案比现有方案好在哪里？

数据阶段

• 数据来自哪里？是否可靠？
• 目标变量是什么？分布如何？
• 有多少缺失值？如何处理？
• 是否有数据泄露风险？

特征工程

• 数值特征是否标准化？
• 类别特征是否编码？（OneHot / Label Encoding）
• 是否需要创建交叉特征？
• 是否需要降维？

模型训练

• Baseline 模型是什么？
• 是否对比了多种算法？
• 是否做了交叉验证？
• 是否用了网格搜索/随机搜索调参？

模型评估

• 评估指标是否与业务目标对齐？
• 是否检查了混淆矩阵？
• 是否做了阈值优化？
• 是否检查了特征重要性（可解释性）？

部署与监控

• 模型是否保存为可加载格式？
• 推理接口是否定义清楚？
• 是否有模型监控方案？（数据漂移、性能衰减）
• 模型多久重新训练一次？

十、经典机器学习系列总结

8 篇文章完整知识链

经典机器学习系列（8 篇）：

回归 分类 无监督
┌──────┐ ┌──────────┐ ┌──────────┐
│ 线性 │ │ 逻辑回归 │──→ 评估 │ 聚类 │
│ 回归 │ │ (02) │ 指标 │ (07) │
│ (01) │ └────┬─────┘ └──────────┘
└──────┘ │
├── KNN (03) 降维
├── SVM (04) ┌──────────┐
├── 决策树 (05) │ PCA/ │
└── 集成学习 (06) │ t-SNE │
├── 随机森林 │ (07) │
├── XGBoost └──────────┘
└── Stacking
│
▼
┌──────────────┐
│ 完整工作流 │
│ 从数据到部署 │
│ (08) │
└──────────────┘

核心能力清单

学完这 8 篇文章，你应该具备：

下一步去哪儿？

经典 ML 学完后，三条进阶路径：

1. 深度学习（你的知识库已有 10 篇笔记）
CNN → RNN → Transformer → 生成模型

2. 特征工程与实操
更多真实数据集练习 → Kaggle 竞赛 → 部署上线

3. 特定领域深入
NLP、计算机视觉、时间序列、推荐系统
（你的知识库中都有对应模块）