EvalRegressionBatchOp extends BatchOperator<EvalRegressionBatchOp> implements EvalRegressionParams<EvalRegressionBatchOp>, EvaluationMetricsCollector<RegressionMetrics, EvalRegressionBatchOp> { @Override public EvalRegressionBatchOp linkFrom(BatchOperator<?>... inputs) { BatchOperator in = checkAndGetFirst(inputs); TableUtil.findColIndexWithAssertAndHint(in.getColNames(), this.getLabelCol()); TableUtil.findColIndexWithAssertAndHint(in.getColNames(), this.getPredictionCol()); TableUtil.assertNumericalCols(in.getSchema(), this.getLabelCol(), this.getPredictionCol()); DataSet<Row> filter = in.select(new String[]{this.getLabelCol(), this.getPredictionCol()}) .getDataSet() .filter(new FilterFunction<Row>() { @Override public boolean filter(Row value) throws Exception { return EvaluationUtil.checkRowFieldNotNull(value); } }); DataSet<Row> out = filter .mapPartition(new CalcLocal()) .withBroadcastSet(DataSetUtil.count(filter), "count") .reduce(new EvaluationUtil.ReduceBaseMetrics()) .flatMap(new EvaluationUtil.SaveDataAsParams()); this.setOutputTable(DataSetConversionUtil.toTable(getMLEnvironmentId(), out, new TableSchema(new String[] {"regression_eval_result"}, new TypeInformation[] {Types.STRING}) )); return this; } EvalRegressionBatchOp();  EvalRegressionBatchOp(Params params); @Override EvalRegressionBatchOp linkFrom(BatchOperator<?>... inputs); @Override RegressionMetrics createMetrics(List<Row> rows);  }
@Test public void test() throws Exception { Row[] data = new Row[] { Row.of(0.4, 0.5), Row.of(0.3, 0.5), Row.of(0.2, 0.6), Row.of(0.6, 0.7), Row.of(0.1, 0.5) }; MemSourceBatchOp input = new MemSourceBatchOp(data, new String[] {"label", "pred"}); RegressionMetrics metrics = new EvalRegressionBatchOp() .setLabelCol("label") .setPredictionCol("pred") .linkFrom(input) .collectMetrics(); Assert.assertEquals(0.275, metrics.getRmse(), 0.01); Assert.assertEquals(-1.56, metrics.getR2(), 0.01); Assert.assertEquals(0.38, metrics.getSse(), 0.01); Assert.assertEquals(141.66, metrics.getMape(), 0.01); Assert.assertEquals(0.24, metrics.getMae(), 0.01); Assert.assertEquals(0.32, metrics.getSsr(), 0.01); Assert.assertEquals(0.14, metrics.getSst(), 0.01); }
@Test public void testEmptyInput() throws Exception{ Row[] rows = new Row[]{ Row.of(null, null), Row.of(1, null), Row.of(null, 1) }; MemSourceBatchOp data = new MemSourceBatchOp(Arrays.asList(rows), new TableSchema(new String[]{"pred", "label"}, new TypeInformation[]{Types.INT, Types.INT})); thrown.expect(Exception.class); EvalRegressionBatchOp op = new EvalRegressionBatchOp(new Params()) .setLabelCol("label") .setPredictionCol("pred") .linkFrom(data); op.print(); }