@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(); }
@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 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 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); }
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); }
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);  }