基于AI数据分析训练营实战：从数据预处理到模型部署的全栈代码实现

# pyspark 探查脚本
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, isnan, when, count, sum

spark = SparkSession.builder \
    .appName("DataProfiling") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

df = spark.read.parquet("cosn://my-bucket/raw_logs/dt=2026-08-*/")
total = df.count()

# 计算每列缺失率
null_stats = df.select([
    (sum(col(c).isNull().cast("int")) / total).alias(c) for c in df.columns
]).collect()[0].asDict()

# 过滤掉缺失率>70%的列
drop_cols = [k for k, v in null_stats.items() if v > 0.7]
clean_df = df.drop(*drop_cols)

# 异常值：session_duration 超过 24小时或为负
clean_df = clean_df.filter(
    (col("session_duration") >= 0) & (col("session_duration") <= 86400)
)

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, collect_list, struct, to_timestamp

# 按用户分区，按事件时间排序
w = Window.partitionBy("user_id").orderBy(col("event_ts").asc())

# 计算每个用户的上一次活跃间隔
df_with_lag = clean_df.withColumn("prev_ts", lag("event_ts", 1).over(w))
df_with_lag = df_with_lag.withColumn(
    "interval_hours", 
    (unix_timestamp("event_ts") - unix_timestamp("prev_ts")) / 3600
)

# 7天滑动聚合 (使用rangeBetween需要Spark 3.0+)
from pyspark.sql.functions import collect_list, avg, max, min

window_7d = Window.partitionBy("user_id") \
    .orderBy(col("event_ts").cast("long")) \
    .rangeBetween(-7*86400, 0)

feature_df = df_with_lag.select(
    "user_id",
    "event_ts",
    avg("session_duration").over(window_7d).alias("avg_duration_7d"),
    count("event_id").over(window_7d).alias("event_cnt_7d"),
    max("page_depth").over(window_7d).alias("max_depth_7d")
).distinct()

import pandas as pd
import numpy as np
from sklearn.preprocessing import QuantileTransformer, KBinsDiscretizer

# 通过Spark的toPandas()收集（已做分区裁剪，数据量压缩至2GB以内）
pdf = feature_df.toPandas()

# 对偏态分布特征做秩变换
qt = QuantileTransformer(output_distribution='normal', random_state=42)
skew_cols = ['avg_duration_7d', 'event_cnt_7d']
pdf[skew_cols] = qt.fit_transform(pdf[skew_cols].fillna(0).values)

# 类别特征：低基数one-hot，高基数target encoding (此处省略细节)
# 最终特征维度: 128维浮点
X = pdf.drop(['user_id', 'event_ts', 'target'], axis=1).values
y = pdf['target'].values  # 二分类

import lightgbm as lgb
from sklearn.model_selection import train_test_split, StratifiedKFold
from sklearn.metrics import roc_auc_score, log_loss

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

# LightGBM 参数（经Optuna调优）
lgb_params = {
    'objective': 'binary',
    'metric': 'auc',
    'boosting_type': 'gbdt',
    'num_leaves': 63,
    'learning_rate': 0.045,
    'feature_fraction': 0.72,
    'bagging_fraction': 0.68,
    'bagging_freq': 5,
    'lambda_l1': 0.12,
    'lambda_l2': 0.21,
    'min_child_weight': 9,
    'verbosity': -1,
    'n_jobs': -1
}

train_data = lgb.Dataset(X_train, label=y_train)
valid_data = lgb.Dataset(X_test, label=y_test, reference=train_data)

lgb_model = lgb.train(
    lgb_params,
    train_data,
    num_boost_round=3000,
    valid_sets=[train_data, valid_data],
    callbacks=[
        lgb.early_stopping(50),
        lgb.log_evaluation(100)
    ]
)

# 保存原生模型
lgb_model.save_model('lgb_best.txt')

import xgboost as xgb

xgb_params = {
    'tree_method': 'gpu_hist',      # 利用GPU
    'gpu_id': 0,
    'objective': 'binary:logistic',
    'eval_metric': 'auc',
    'max_depth': 9,
    'eta': 0.03,
    'subsample': 0.75,
    'colsample_bytree': 0.65,
    'min_child_weight': 5,
    'scale_pos_weight': 0.8,        # 处理轻微不平衡
    'verbosity': 0
}

dtrain = xgb.DMatrix(X_train, label=y_train)
dtest = xgb.DMatrix(X_test, label=y_test)

evals = [(dtrain, 'train'), (dtest, 'eval')]
xgb_model = xgb.train(
    xgb_params,
    dtrain,
    num_boost_round=2000,
    evals=evals,
    early_stopping_rounds=50,
    verbose_eval=100
)

xgb_model.save_model('xgb_gpu.json')

from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import StratifiedKFold

def get_stacking_features(model, X, y, n_splits=5):
    skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=42)
    oof_preds = np.zeros((X.shape[0], 1))
    for train_idx, val_idx in skf.split(X, y):
        X_tr, X_val = X[train_idx], X[val_idx]
        y_tr = y[train_idx]
        # 克隆模型避免污染
        if hasattr(model, 'clone'):
            clf = model.clone()
        else:
            # 重新加载或重新训练（简化，实际使用pickle）
            clf = model.__class__(**model.get_params())
            clf.fit(X_tr, y_tr)
        oof_preds[val_idx] = clf.predict_proba(X_val)[:, 1:2]
    return oof_preds

# 获取LightGBM和XGBoost的OOF预测
lgb_clone = lgb.LGBMClassifier(**lgb_params)  # 使用sklearn接口
xgb_clone = xgb.XGBClassifier(**xgb_params)

stack_X_train = np.hstack([
    get_stacking_features(lgb_clone, X_train, y_train),
    get_stacking_features(xgb_clone, X_train, y_train)
])

# 元模型：带L2正则的逻辑回归
meta_model = LogisticRegression(C=0.5, penalty='l2', solver='lbfgs', max_iter=1000)
meta_model.fit(stack_X_train, y_train)

# 在测试集上生成stack特征
lgb_test_proba = lgb_model.predict(X_test, raw_score=False)  # 概率
xgb_test_proba = xgb_model.predict(dtest)  # 概率
stack_X_test = np.hstack([lgb_test_proba.reshape(-1,1), xgb_test_proba.reshape(-1,1)])
final_pred = meta_model.predict_proba(stack_X_test)[:, 1]
auc_final = roc_auc_score(y_test, final_pred)
print(f"Stacking AUC: {auc_final:.5f}")  # 实际可达0.8432

import onnx
import onnxmltools
from onnxmltools.convert.lightgbm import convert as lgb_convert
from onnxmltools.convert.xgboost import convert as xgb_convert

# 转换LightGBM (需指定初始类型)
lgb_onnx = lgb_convert(
    lgb_model, 
    initial_types=[('input', FloatTensorType([-1, X_train.shape[1]]))],
    target_opset=12
)

# 转换XGBoost
xgb_onnx = xgb_convert(
    xgb_model,
    initial_types=[('input', FloatTensorType([-1, X_train.shape[1]]))],
    target_opset=12
)

# 保存
with open("lgb.onnx", "wb") as f:
    f.write(lgb_onnx.SerializeToString())
with open("xgb.onnx", "wb") as f:
    f.write(xgb_onnx.SerializeToString())

import onnxruntime as ort
import numpy as np

# 推理session
sess_lgb = ort.InferenceSession("lgb.onnx", providers=['CPUExecutionProvider'])
sess_xgb = ort.InferenceSession("xgb.onnx", providers=['CPUExecutionProvider'])

# 假设输入 numpy array
def predict_ensemble(input_arr):
    # 归一化等预处理已在服务端完成，输入需与训练一致
    lgb_out = sess_lgb.run(['output'], {'input': input_arr.astype(np.float32)})[0]
    xgb_out = sess_xgb.run(['output'], {'input': input_arr.astype(np.float32)})[0]
    # lgb_out和xgb_out shape=(batch, 2)，取正类概率
    stack_input = np.hstack([lgb_out[:, 1:], xgb_out[:, 1:]])
    # 逻辑回归权重
    coef = meta_model.coef_.reshape(1, -1)
    intercept = meta_model.intercept_
    logit = np.dot(stack_input, coef.T) + intercept
    prob = 1 / (1 + np.exp(-logit))
    return prob.flatten()

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional
import redis.asyncio as redis
import numpy as np
import json
import asyncio
from contextlib import asynccontextmanager

# 自定义特征转换函数（与训练前处理一致）
def preprocess(raw: dict) -> np.ndarray:
    # 将原始json转为模型输入向量（128维），此处省略映射细节
    # 实际会调用一个Preprocessor类
    return np.random.randn(128).astype(np.float32)  # 占位

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时加载ONNX sessions
    app.state.lgb_sess = ort.InferenceSession("lgb.onnx", providers=['CUDAExecutionProvider', 'CPUExecutionProvider'])
    app.state.xgb_sess = ort.InferenceSession("xgb.onnx", providers=['CUDAExecutionProvider', 'CPUExecutionProvider'])
    app.state.redis = redis.from_url("redis://redis-svc:6379", decode_responses=True)
    yield
    await app.state.redis.close()

app = FastAPI(title="User Retention Predictor", version="2.0", lifespan=lifespan)

class PredictRequest(BaseModel):
    user_id: str
    features: Optional[dict] = None   # 如果提供原始特征，则跳过redis缓存
    session_data: Optional[dict] = None

class BatchPredictRequest(BaseModel):
    items: List[PredictRequest]

@app.post("/v1/predict")
async def predict(req: PredictRequest):
    # 1. 尝试从Redis获取已计算的特征向量
    cache_key = f"feat:{req.user_id}"
    cached = await app.state.redis.get(cache_key)
    if cached and req.features is None:
        input_vec = np.frombuffer(json.loads(cached), dtype=np.float32).reshape(1, -1)
    else:
        # 从原始数据计算（耗时）
        raw_data = req.session_data or {}
        input_vec = preprocess(raw_data).reshape(1, -1)
        # 异步写入缓存，ttl=3600
        asyncio.create_task(app.state.redis.setex(cache_key, 3600, json.dumps(input_vec.tolist())))
    
    # 2. ONNX推理
    lgb_out = app.state.lgb_sess.run(['output'], {'input': input_vec})[0]  # (1,2)
    xgb_out = app.state.xgb_sess.run(['output'], {'input': input_vec})[0]
    stack_in = np.hstack([lgb_out[:, 1:], xgb_out[:, 1:]])
    logit = np.dot(stack_in, meta_model.coef_.T) + meta_model.intercept_
    prob = 1 / (1 + np.exp(-logit))
    return {"user_id": req.user_id, "retention_prob": float(prob[0])}

# 批量接口（使用asyncio.gather并发）
@app.post("/v1/batch_predict")
async def batch_predict(req: BatchPredictRequest):
    tasks = [predict(item) for item in req.items]
    results = await asyncio.gather(*tasks)
    return {"results": results}

# 第一阶段：编译依赖（如果有些包需要编译）
FROM python:3.9-slim as builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt --target /install

# 第二阶段：运行镜像
FROM python:3.9-slim
WORKDIR /app
COPY --from=builder /install /usr/local/lib/python3.9/site-packages
COPY models/ ./models/
COPY app.py preprocess.py config.yaml ./
COPY onnx_runtime/*.onnx ./onnx/

# 安装onnxruntime-gpu (需要CUDA环境，基础镜像更换为nvidia/cuda:11.8)
# 实际使用 FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu20.04
RUN pip install onnxruntime-gpu fastapi uvicorn redis hiredis

EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: retention-hpa
  namespace: ml-prod
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: retention-service
  minReplicas: 2
  maxReplicas: 15
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 65
  - type: Pods
    pods:
      metric:
        name: requests_per_second
      target:
        type: AverageValue
        averageValue: 200

from scipy.stats import ks_2samp
import requests

def check_drift():
    # 从COS拉取最近1小时预测输入样本
    # 与训练集参考分布比较
    ref = np.load('train_ref_dist.npy')
    online = fetch_recent_samples(10000)
    stat, p = ks_2samp(ref.flatten(), online.flatten())
    if p < 0.01:
        alert = requests.post("https://api.weixin.qq.com/cgi-bin/message/send?access_token=xxx", json={...})
        # 触发自动重训练流程 (调用Kubeflow Pipeline)