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

功能介绍

  • Esd算法又叫做箱线图算法, 是一种常用的异常检测算法.

    参数说明

    | 名称 | 中文名称 | 描述 | 类型 | 是否必须？ | 取值范围 | 默认值 | | —- | —- | —- | —- | —- | —- | —- |

| precedingTime | 时间窗口大小 | 时间窗口大小 | String | ✓ | | |

| predictionCol | 预测结果列名 | 预测结果列名 | String | ✓ | | |

| alpha | 置信度 | 置信度 | Double | | | 0.05 |

| direction | Not available! | Not available! | String | | “POSITIVE”, “NEGATIVE”, “BOTH” | “BOTH” |

| featureCol | 特征列名 | 特征列名，默认选最左边的列 | String | | 所选列类型为 [BIGDECIMAL, BIGINTEGER, BYTE, DOUBLE, FLOAT, INTEGER, LONG, SHORT] | null |

| groupCols | 分组列名数组 | 分组列名，多列，可选，默认不选 | String[] | | | null |

| maxIter | 最大迭代步数 | 最大迭代步数 | Integer | | | |

| outlierThreshold | 异常评分阈值 | 只有评分大于该阈值才会被认为是异常点 | Double | | | |

| precedingRows | 数据窗口大小 | 数据窗口大小 | Integer | | | null |

| predictionDetailCol | 预测详细信息列名 | 预测详细信息列名 | String | | | |

| timeCol | 时间戳列(TimeStamp) | 时间戳列(TimeStamp) | String | | | null |

| numThreads | 组件多线程线程个数 | 组件多线程线程个数 | Integer | | | 1 |

代码示例

Python 代码

  1. from pyalink.alink import *
  3. import pandas as pd
  5. useLocalEnv(1)
  7. import time, datetime
  8. import numpy as np
  9. import pandas as pd
  11. data = pd.DataFrame([
  12.             [1, datetime.datetime.fromtimestamp(1), 10.0, 0],
  13.             [1, datetime.datetime.fromtimestamp(2), 11.0, 0],
  14.             [1, datetime.datetime.fromtimestamp(3), 12.0, 0],
  15.             [1, datetime.datetime.fromtimestamp(4), 13.0, 0],
  16.             [1, datetime.datetime.fromtimestamp(5), 14.0, 0],
  17.             [1, datetime.datetime.fromtimestamp(6), 15.0, 0],
  18.             [1, datetime.datetime.fromtimestamp(7), 16.0, 0],
  19.             [1, datetime.datetime.fromtimestamp(8), 17.0, 0],
  20.             [1, datetime.datetime.fromtimestamp(9), 18.0, 0],
  21.             [1, datetime.datetime.fromtimestamp(10), 19.0, 0]
  22. ])
  24. dataOp = dataframeToOperator(data, schemaStr='id int, ts timestamp, val double, label int', op_type='stream')
  26. outlierOp = EsdOutlierStreamOp()\
  27.             .setGroupCols(["id"])\
  28.             .setTimeCol("ts")\
  29.             .setPrecedingRows(3)\
  30.             .setFeatureCol("val")\
  31.             .setPredictionCol("pred")\
  32.             .setPredictionDetailCol("pred_detail")
  34. dataOp.link(outlierOp).print()
  36. StreamOperator.execute()

Java 代码

  1. package com.alibaba.alink.operator.stream.outlier;
  3. import org.apache.flink.types.Row;
  5. import com.alibaba.alink.operator.stream.StreamOperator;
  6. import com.alibaba.alink.operator.stream.sink.CollectSinkStreamOp;
  7. import com.alibaba.alink.operator.stream.source.MemSourceStreamOp;
  8. import junit.framework.TestCase;
  9. import org.junit.Assert;
  10. import org.junit.Test;
  12. import java.sql.Timestamp;
  13. import java.util.Arrays;
  14. import java.util.List;
  16. public class EsdOutlierStreamOpTest extends TestCase {
  17.     @Test
  18.     public void test() throws Exception {
  19.         List <Row> mTableData = Arrays.asList(
  20.             Row.of(1, new Timestamp(1), 10.0, 0),
  21.             Row.of(1, new Timestamp(2), 11.0, 0),
  22.             Row.of(1, new Timestamp(3), 12.0, 0),
  23.             Row.of(1, new Timestamp(4), 13.0, 0),
  24.             Row.of(1, new Timestamp(5), 14.0, 0),
  25.             Row.of(1, new Timestamp(6), 15.0, 0),
  26.             Row.of(1, new Timestamp(7), 16.0, 0),
  27.             Row.of(1, new Timestamp(8), 17.0, 0),
  28.             Row.of(1, new Timestamp(9), 18.0, 0),
  29.             Row.of(1, new Timestamp(10), 19.0, 0)
  30.         );
  32.         MemSourceStreamOp dataOp = new MemSourceStreamOp(mTableData, new String[] {"id", "ts", "val", "label"});
  34.         EsdOutlierStreamOp outlierOp = new EsdOutlierStreamOp()
  35.             .setGroupCols("id")
  36.             .setTimeCol("ts")
  37.             .setPrecedingRows(3)
  38.             .setFeatureCol("val")
  39.             .setPredictionCol("pred")
  40.             .setPredictionDetailCol("pred_detail");
  42.         dataOp.link(outlierOp).print();
  44.         StreamOperator.execute();
  45.     }
  46. }

运行结果

无