AI工业数据挖掘工具实战：从数据清洗到模型部署的全流程优化

1次阅读

共计 4915 个字符，预计需要花费 13 分钟才能阅读完成。

工业数据挖掘与常规数据分析相比，面临着独特的挑战。这些挑战直接影响到数据挖掘的效果和实际应用价值。

数据质量问题
传感器噪声：工业设备传感器常受到电磁干扰、机械振动等影响，导致采集数据包含大量噪声
数据缺失：设备维护、网络中断等情况造成数据不完整
采样频率不一致：不同设备的数据采集周期可能不同
数据结构复杂
多源异构数据：同时包含时序数据、图像数据、结构化日志等
高维度特征：单个设备可能产生数百个监测指标
实时性要求高
预测性维护场景需要毫秒级响应
在线质量控制要求低延迟推理

面对工业数据的这些特点，我们对比了主流数据处理框架：

Spark MLlib 优势
分布式计算能力，适合处理 TB 级工业数据
内置工业常用算法（如 Isolation Forest 异常检测）
与 Hadoop 生态无缝集成
Scikit-learn 优势
算法丰富度更高
更灵活的模型调参接口
更成熟的模型持久化方案

最终选择 PySpark+Scikit-learn 组合的原因：
1. 利用 PySpark 处理数据预处理和特征工程阶段的分布式计算需求
2. 在模型训练阶段切换到 Scikit-learn 获得更优的算法效果
3. 通过 Py4J 实现两者内存数据高效交换

from pyspark.sql import functions as F

# 缺失值处理（分布式执行）def handle_missing(df):
    # 数值型字段用中位数填充
    numeric_cols = [field.name for field in df.schema.fields 
                   if isinstance(field.dataType, (IntegerType, DoubleType, FloatType))]
    medians = df.approxQuantile(numeric_cols, [0.5], 0.01)

    for col, median in zip(numeric_cols, medians):
        df = df.withColumn(col, F.when(F.col(col).isNull(), median).otherwise(F.col(col)))

    # 类别型字段用众数填充
    string_cols = [...]  # 获取类别型字段列表
    modes = {}
    for col in string_cols:
        mode = df.groupBy(col).count().orderBy("count", ascending=False).first()[0]
        modes[col] = mode

    for col in string_cols:
        df = df.withColumn(col, F.when(F.col(col).isNull(), modes[col]).otherwise(F.col(col)))

    return df

# 异常值检测（基于 IQR 方法）def detect_outliers(df, col):
    quantiles = df.approxQuantile(col, [0.25, 0.75], 0.05)
    iqr = quantiles[1] - quantiles[0]
    bounds = [quantiles[0] - 1.5*iqr, quantiles[1] + 1.5*iqr]

    return df.withColumn(f"{col}_outlier", 
        ~F.col(col).between(bounds[0], bounds[1])
    )

from sklearn.feature_selection import mutual_info_classif
from sklearn.decomposition import PCA

# 互信息特征选择
def select_by_mi(X, y, threshold=0.05):
    """
    X: pandas DataFrame 格式的特征矩阵
    y: 目标变量
    threshold: 互信息阈值
    """
    mi_scores = mutual_info_classif(X, y)
    selected = X.columns[mi_scores > threshold]
    return X[selected]

# PCA 降维（在特征选择后执行）def apply_pca(X, n_components=0.95):
    """n_components: 保留的方差比例"""
    pca = PCA(n_components=n_components, svd_solver='full')
    X_pca = pca.fit_transform(X)
    print(f"原始维度: {X.shape[1]}, 降维后: {X_pca.shape[1]}")
    return X_pca

import xgboost as xgb
from sklearn.model_selection import RandomizedSearchCV

# 轻量级参数配置
base_params = {
    'n_estimators': 100,
    'max_depth': 6,
    'learning_rate': 0.1,
    'subsample': 0.8,
    'colsample_bytree': 0.8,
    'objective': 'binary:logistic',
    'n_jobs': -1
}

# 参数搜索空间
param_dist = {'max_depth': [3, 4, 5, 6],
    'learning_rate': [0.01, 0.05, 0.1],
    'gamma': [0, 0.1, 0.2],
    'reg_alpha': [0, 0.1, 1],
    'reg_lambda': [0, 0.1, 1]
}

# 使用贝叶斯优化替代网格搜索
from skopt import BayesSearchCV

opt = BayesSearchCV(xgb.XGBClassifier(**base_params),
    param_dist,
    n_iter=30,
    cv=3,
    scoring='roc_auc',
    verbose=1
)
opt.fit(X_train, y_train)

import joblib
import json

# 保存模型和元数据
def save_model(model, preprocessor, metadata, path):
    """
    model: 训练好的模型对象
    preprocessor: 包含 PCA 等预处理步骤的 pipeline
    metadata: 特征列表等元信息
    path: 保存路径
    """
    # 使用 joblib 替代 pickle 以获得更好的性能
    joblib.dump({
        'model': model,
        'preprocessor': preprocessor
    }, f"{path}/model.joblib")

    # 单独保存元数据
    with open(f"{path}/metadata.json", 'w') as f:
        json.dump(metadata, f)

# 加载时验证版本兼容性
def load_model(path, expected_sklearn_version='1.0.2'):
    import sklearn
    if sklearn.__version__ != expected_sklearn_version:
        print(f"Warning: 模型使用 sklearn {expected_sklearn_version}训练，当前版本{sklearn.__version__}")

    return joblib.load(f"{path}/model.joblib")

from flask import Flask, request, jsonify
import gunicorn

app = Flask(__name__)

# 全局加载模型
model_pkg = load_model('./model')

@app.route('/predict', methods=['POST'])
def predict():
    # 输入数据验证
    try:
        data = request.get_json()
        features = preprocess_input(data['features'])  # 自定义预处理
    except Exception as e:
        return jsonify({'error': str(e)}), 400

    # 批量预测更高效
    if isinstance(features, list):
        preds = model_pkg['model'].predict_proba(features)
        return jsonify({'predictions': preds[:, 1].tolist()})
    else:
        pred = model_pkg['model'].predict_proba([features])[0,1]
        return jsonify({'prediction': float(pred)})

# Gunicorn 配置建议
"""
# gunicorn_config.py
workers = min(4, (os.cpu_count() or 1) + 1)
worker_class = 'gevent'
keepalive = 5
timeout = 30
preload_app = True  # 避免拷贝模型内存
"""

时间序列数据分割
必须按时间划分训练 / 测试集
避免随机分割导致未来信息泄露
特征工程中的陷阱
全局统计量（如均值、标准差）应该在每个时间窗口内单独计算
避免使用包含未来信息的滚动窗口
代码示例：安全的分割方法

from sklearn.model_selection import TimeSeriesSplit

tscv = TimeSeriesSplit(n_splits=5)
for train_index, test_index in tscv.split(X):
    X_train, X_test = X.iloc[train_index], X.iloc[test_index]
    y_train, y_test = y.iloc[train_index], y.iloc[test_index]

    # 确保测试集时间都在训练集之后
    assert X_test.index.min() > X_train.index.max()

PySpark 内存配置

# 提交任务时设置
spark-submit --executor-memory 8G \
             --driver-memory 4G \
             --conf spark.memory.fraction=0.8 \
             --conf spark.memory.storageFraction=0.3

Scikit-learn 内存优化技巧
对于大型矩阵，使用 np.float32 替代np.float64

使用 memory 参数缓存管道步骤

from sklearn.pipeline import Pipeline
from sklearn.externals import joblib

pipe = Pipeline([('preprocessor', preprocessor),
    ('model', model)
], memory=joblib.Memory('./cache'))

我们在真实工业数据集（约 200 万条记录，300+ 特征）上进行了测试：