@Override
  public void transform(Row row, PointCollector collector) throws Exception {
    double value = Util.getValueAsDouble(row);
    long timestamp = row.getTime();
    if (Double.isFinite(value) && !Double.isNaN(value)) {
      if (v.isFull()) {
        final double frontValue = Double.parseDouble(v.pop().toString());
        switch (dataType) {
          case INT32:
            v.push(row.getInt(0));
            break;
          case INT64:
            v.push(row.getLong(0));
            break;
          case DOUBLE:
            v.push(row.getDouble(0));
            break;
          case FLOAT:
            v.push(row.getFloat(0));
            break;
          case TIMESTAMP:
          case DATE:
          case TEXT:
          case STRING:
          case BLOB:
          case BOOLEAN:
          default:
            break;
        }
        t.pop();
        t.push(timestamp);
        this.sumX1 = this.sumX1 - frontValue + value;
        this.sumX2 = this.sumX2 - frontValue * frontValue + value * value;
        this.mean = this.sumX1 / v.getSize();
        this.variance = this.sumX2 / v.getSize() - this.mean * this.mean;
        if (Math.abs(value - mean)
            > multipleK * Math.sqrt(this.variance * v.getSize() / (v.getSize() - 1))) {
          Util.putValue(collector, dataType, timestamp, Util.getValueAsObject(row));
        }
      } else {
        switch (dataType) {
          case INT32:
            v.push(row.getInt(0));
            break;
          case INT64:
            v.push(row.getLong(0));
            break;
          case DOUBLE:
            v.push(row.getDouble(0));
            break;
          case FLOAT:
            v.push(row.getFloat(0));
            break;
          case BLOB:
          case BOOLEAN:
          case STRING:
          case TEXT:
          case DATE:
          case TIMESTAMP:
          default:
            break;
        }
        t.push(timestamp);
        this.sumX1 = this.sumX1 + value;
        this.sumX2 = this.sumX2 + value * value;
        this.mean = this.sumX1 / v.getSize();
        this.variance = this.sumX2 / v.getSize() - this.mean * this.mean;
        if (v.getSize() == this.windowSize) {
          double stddev = Math.sqrt(this.variance * v.getSize() / (v.getSize() - 1));
          for (int i = 0; i < v.getSize(); i++) {
            Object vi = this.v.get(i);
            timestamp = this.t.get(i);
            if (Math.abs(Double.parseDouble(vi.toString()) - mean) > multipleK * stddev) {
              Util.putValue(collector, dataType, timestamp, vi);
            }
          }
        }
      }
    }
  }
This function detects outliers which lies over average +/- k * sigma.
library-udf/src/main/java/org/apache/iotdb/library/anomaly/UDTFKSigma.java