基于ML.NET检测DataStage ETL作业运行异常的后续步骤咨询

1. 特征工程与数据预处理 #

首先需要构建能反映作业运行效率的核心特征，同时清洗无效数据：

• 计算单位行耗时特征：针对消耗行数和生成行数分别计算，比如 SecPerConsumedRow = ElapsedRunSecs / TotalRowsConsumed、SecPerProducedRow = ElapsedRunSecs / TotalRowsProduced，这两个特征直接体现作业处理每行的效率，是异常检测的核心依据。
• 提取时间维度特征：从RunStartTimestamp中提取小时、星期几、是否工作日等特征，部分作业可能在特定时段（比如凌晨资源充足）运行更快，这些特征能帮助模型区分正常的时段差异和异常的效率下降。
• 过滤无效数据：剔除TotalRowsConsumed或TotalRowsProduced为0的记录（无数据处理的作业无参考意义），同时处理ElapsedRunSecs为0的异常数据。

用ML.NET实现特征工程的示例代码：

// 定义包含新特征的类
public class JobRunWithFeatures
{
    public float ElapsedRunSecs { get; set; }
    public float TotalRowsConsumed { get; set; }
    public float TotalRowsProduced { get; set; }
    public float SecPerConsumedRow { get; set; }
    public float SecPerProducedRow { get; set; }
    public int HourOfDay { get; set; }
    public int DayOfWeek { get; set; }
}

// 构建特征工程管道
var pipeline = mlContext.Transforms.CustomMapping<JobRun, JobRunWithFeatures>(
    (input, output) =>
    {
        // 计算单位行耗时（避免除0，加极小值）
        output.SecPerConsumedRow = input.TotalRowsConsumed == 0 ? 0 : input.ElapsedRunSecs / (input.TotalRowsConsumed + float.Epsilon);
        output.SecPerProducedRow = input.TotalRowsProduced == 0 ? 0 : input.ElapsedRunSecs / (input.TotalRowsProduced + float.Epsilon);
        // 提取时间特征
        output.HourOfDay = input.RunStartTimestamp.Hour;
        output.DayOfWeek = (int)input.RunStartTimestamp.DayOfWeek;
        // 保留原始字段
        output.ElapsedRunSecs = input.ElapsedRunSecs;
        output.TotalRowsConsumed = input.TotalRowsConsumed;
        output.TotalRowsProduced = input.TotalRowsProduced;
    }, contractName: null)
// 过滤无效数据
.Append(mlContext.Data.FilterRowsByColumn("TotalRowsConsumed", lowerBound: 1))
.Append(mlContext.Data.FilterRowsByColumn("TotalRowsProduced", lowerBound: 1));

// 应用管道到训练集和测试集
var transformedTrainData = pipeline.Fit(testTrainSplit.TrainSet).Transform(testTrainSplit.TrainSet);
var transformedTestData = pipeline.Transform(testTrainSplit.TestSet);

2. 选择并训练异常检测模型 #

针对需求推荐两种落地方案：

方案A：基于回归的残差异常检测 #

先训练回归模型预测正常作业的单位行耗时，再计算实际值与预测值的残差，残差超过阈值的即为异常：

// 定义回归预测结果类
public class RegressionPrediction
{
    public float Score { get; set; } // 预测的单位行耗时
    public float SecPerConsumedRow { get; set; } // 实际单位行耗时
    public float ElapsedRunSecs { get; set; }
    public float TotalRowsConsumed { get; set; }
    public DateTime RunStartTimestamp { get; set; }
}

// 构建回归模型管道
var regressionPipeline = mlContext.Transforms.Concatenate("Features", "TotalRowsConsumed", "TotalRowsProduced", "HourOfDay", "DayOfWeek")
.Append(mlContext.Regression.Trainers.Sdca(labelColumnName: "SecPerConsumedRow", maximumNumberOfIterations: 100));

// 训练模型
var regressionModel = regressionPipeline.Fit(transformedTrainData);

// 预测并计算残差
var predictions = regressionModel.Transform(transformedTestData);

方案B：无监督异常检测算法（Isolation Forest） #

直接使用ML.NET的Isolation Forest算法检测异常点，适合无标签数据场景：

// 定义异常检测结果类
public class AnomalyPredictionResult
{
    public bool PredictedLabel { get; set; } // 1=异常，0=正常
    public float Score { get; set; } // 异常分数，值越大异常程度越高
    public float ElapsedRunSecs { get; set; }
    public float TotalRowsConsumed { get; set; }
    public DateTime RunStartTimestamp { get; set; }
}

// 构建异常检测管道
var anomalyPipeline = mlContext.Transforms.Concatenate("Features", "SecPerConsumedRow", "SecPerProducedRow", "HourOfDay", "DayOfWeek")
.Append(mlContext.AnomalyDetection.Trainers.IsolationForest(contamination: 0.05)); // contamination设为预期异常比例，比如5%

// 训练模型
var anomalyModel = anomalyPipeline.Fit(transformedTrainData);

// 预测异常
var anomalyPredictions = anomalyModel.Transform(transformedTestData);

3. 模型评估与阈值确定 #

• 回归残差方案：统计残差的分布，取95分位数作为阈值，残差超过该值的标记为异常；同时可通过回归指标评估模型拟合效果：

var metrics = mlContext.Regression.Evaluate(predictions, labelColumnName: "SecPerConsumedRow");
Console.WriteLine($"R2 Score: {metrics.RSquared}");
Console.WriteLine($"RMSE: {metrics.RootMeanSquaredError}");

• Isolation Forest方案：通过测试集调整contamination参数，或根据业务需求调整异常分数阈值，平衡误报率和漏报率。

4. 异常检测与结果输出 #

将训练好的模型应用到新的作业运行数据上，标记并输出异常作业信息：

// 加载新的作业运行数据（示例）
var newJobRuns = new List<JobRun> { /* 填充新数据 */ };
var newDataView = mlContext.Data.LoadFromEnumerable(newJobRuns);
var transformedNewData = pipeline.Transform(newDataView);
var newPredictions = anomalyModel.Transform(transformedNewData);

// 转换为可枚举对象查看结果
var results = mlContext.Data.CreateEnumerable<AnomalyPredictionResult>(newPredictions, reuseRowObject: false);
foreach (var result in results)
{
    if (result.PredictedLabel)
    {
        Console.WriteLine($"异常作业：开始时间{result.RunStartTimestamp}，消耗行数{result.TotalRowsConsumed}，耗时{result.ElapsedRunSecs}秒，异常分数{result.Score}");
    }
}

内容的提问来源于stack exchange，提问作者Mihaimyh