异常检测 - IForest序列异常检测 (IForestOutlier4GroupedDataStreamOp) - 《Alink 1.5.6 文档 - 帮助手册 - 教程》

功能介绍
- 文献或出处
参数说明
代码示例

Java 类名：com.alibaba.alink.operator.stream.outlier.IForestOutlier4GroupedDataStreamOp
Python 类名：IForestOutlier4GroupedDataStreamOp

功能介绍

iForest 可以识别数据中异常点，在异常检测领域有比较好的效果。算法使用 sub-sampling 方法，降低了算法的计算复杂度。

文献或出处

Isolation Forest
参数说明
| 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

代码示例

Python 代码

import time
import datetime
import numpy as np
import pandas as pd
data = pd.DataFrame([
    [1, datetime.datetime.fromtimestamp(1), 10.0, 0],
    [1, datetime.datetime.fromtimestamp(2), 11.0, 0],
    [1, datetime.datetime.fromtimestamp(3), 12.0, 0],
    [1, datetime.datetime.fromtimestamp(4), 13.0, 0],
    [1, datetime.datetime.fromtimestamp(5), 14.0, 0],
    [1, datetime.datetime.fromtimestamp(6), 15.0, 0],
    [1, datetime.datetime.fromtimestamp(7), 16.0, 0],
    [1, datetime.datetime.fromtimestamp(8), 17.0, 0],
    [1, datetime.datetime.fromtimestamp(9), 18.0, 0],
    [1, datetime.datetime.fromtimestamp(10), 19.0, 0]
])
dataOp = dataframeToOperator(
    data, schemaStr='id int, ts timestamp, val double, label int', op_type='stream')
dataOp.link(
    OverCountWindowStreamOp()
    .setGroupCols(["id"])
    .setTimeCol("ts")
    .setPrecedingRows(5)
    .setClause("MTABLE_AGG(ts, val) as series_data")
    .setReservedCols(["id", "label"])
).link(
    IForestOutlier4GroupedDataStreamOp()
    .setInputMTableCol("series_data")
    .setFeatureCols(["val"])
    .setOutputMTableCol("output_series")
    .setPredictionCol("pred")
).link(
    FlattenMTableStreamOp()
    .setSelectedCol("output_series")
    .setSchemaStr("ts TIMESTAMP, val DOUBLE, pred BOOLEAN")
).print()
StreamOperator.execute()

Java 代码

import org.apache.flink.types.Row;
import com.alibaba.alink.operator.stream.StreamOperator;
import com.alibaba.alink.operator.stream.dataproc.FlattenMTableStreamOp;
import com.alibaba.alink.operator.stream.feature.OverCountWindowStreamOp;
import com.alibaba.alink.operator.stream.outlier.IForestOutlier4GroupedDataStreamOp;
import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
import org.junit.Test;
import java.sql.Timestamp;
import java.util.Arrays;
import java.util.List;
public class IForestOutlier4GroupedDataStreamOpTest {
    @Test
    public void test() throws Exception {
        List <Row> mTableData = Arrays.asList(
            Row.of(1, new Timestamp(1), 10.0, 0),
            Row.of(1, new Timestamp(2), 11.0, 0),
            Row.of(1, new Timestamp(3), 12.0, 0),
            Row.of(1, new Timestamp(4), 13.0, 0),
            Row.of(1, new Timestamp(5), 14.0, 0),
            Row.of(1, new Timestamp(6), 15.0, 0),
            Row.of(1, new Timestamp(7), 16.0, 0),
            Row.of(1, new Timestamp(8), 17.0, 0),
            Row.of(1, new Timestamp(9), 18.0, 0),
            Row.of(1, new Timestamp(10), 19.0, 0)
        );
        MemSourceStreamOp dataOp = new MemSourceStreamOp(mTableData,
            new String[] {"id", "ts", "val", "label"});
        dataOp.link(
            new OverCountWindowStreamOp()
                .setGroupCols("id")
                .setTimeCol("ts")
                .setPrecedingRows(10)
                .setClause("MTABLE_AGG(ts, val) as series_data")
                .setReservedCols("id", "label")
        ).link(
            new IForestOutlier4GroupedDataStreamOp()
                .setInputMTableCol("series_data")
                .setFeatureCols("val")
                .setOutputMTableCol("output_series")
                .setPredictionCol("pred")
        ).link(
            new FlattenMTableStreamOp()
                .setSelectedCol("output_series")
                .setSchemaStr("ts TIMESTAMP, val DOUBLE, pred BOOLEAN")
        ).print();
        StreamOperator.execute();
    }
}

运行结果

| id | label | series_data | output_series | ts | val | pred | | —- | —- | —- | —- | —- | —- | —- |

| 1 | 0 | MTable(1,2)(ts,val) | MTable(1,3)(ts,val,pred) | 1970-01-01 08:00:00.001 | 10.0000 | false |