IsotonicRegTrainBatchOp extends BatchOperator<IsotonicRegTrainBatchOp> implements IsotonicRegTrainParams<IsotonicRegTrainBatchOp> { @Override public IsotonicRegTrainBatchOp linkFrom(BatchOperator<?>... inputs) { BatchOperator<?> in = checkAndGetFirst(inputs); String labelColName = this.getLabelCol(); String featureColName = this.getFeatureCol(); String weightColName = this.getWeightCol(); String vectorColName = this.getVectorCol(); boolean isotonic = this.getIsotonic(); int index = this.getFeatureIndex(); String[] selectedColNames; if (null == vectorColName && null != featureColName) { if (weightColName == null) { selectedColNames = new String[]{labelColName, featureColName}; } else { selectedColNames = new String[]{labelColName, featureColName, weightColName}; } } else if (null == featureColName && null != vectorColName) { if (weightColName == null) { selectedColNames = new String[]{labelColName, vectorColName}; } else { selectedColNames = new String[]{labelColName, vectorColName, weightColName}; } } else { throw new IllegalArgumentException("Either featureColName or vectorColName is required!"); } DataSet<Tuple3<Double, Double, Double>> dataSet = in.select(selectedColNames) .getDataSet() .map(new MapFunction<Row, Tuple3<Double, Double, Double>>() { @Override public Tuple3<Double, Double, Double> map(Row row) { double label = ((Number) row.getField(0)).doubleValue(); label = isotonic ? label : -label; double feature = null == vectorColName ? ((Number) row.getField(1)).doubleValue() : VectorUtil.getVector(row.getField(1)).get(index); double weight = null == weightColName ? 1.0 : ((Number) row.getField(2)).doubleValue(); if (weight < 0) { throw new IllegalArgumentException("Weights must be non-negative!"); } return Tuple3.of(label, feature, weight); } } ); DataSet<byte[]> model = dataSet .filter(new FilterFunction<Tuple3<Double, Double, Double>>() { @Override public boolean filter(Tuple3<Double, Double, Double> value) { return value.f2 > 0; } }) .rebalance() .partitionByRange(1) .mapPartition(new PoolAdjacentViolators()); DataSet<Row> res = model .mapPartition(new BuildModel(isotonic, featureColName, vectorColName, index)) .setParallelism(1); this.setOutput(res, new IsotonicRegressionConverter().getModelSchema()); return this; } IsotonicRegTrainBatchOp();  IsotonicRegTrainBatchOp(Params params); @Override IsotonicRegTrainBatchOp linkFrom(BatchOperator<?>... inputs);  }
@Test public void isotonicRegTest() throws Exception { int length = 15; Object[][] inTrain=new Object[length][3]; for (int i=0;i<length;++i){ inTrain[i][0]=tuples[i].f0; inTrain[i][1]=tuples[i].f1; inTrain[i][2]=tuples[i].f2; } Object[][] inTest=new Object[10][3]; for (int i=0;i<10;++i){ inTest[i][0]=tupleTest[i].f0; inTest[i][1]=tupleTest[i].f1; inTest[i][2]=tupleTest[i].f2; } String[] colNames=new String[]{"col1","col2","col3"}; MemSourceBatchOp trainData=new MemSourceBatchOp(inTrain,colNames); MemSourceBatchOp predictData = new MemSourceBatchOp(inTest, new String[]{"col1","col2","col3"}); IsotonicRegTrainBatchOp model=new IsotonicRegTrainBatchOp().setLabelCol("col1").setFeatureCol("col2").setWeightCol("col3") .linkFrom(trainData); model.print(); IsotonicRegPredictBatchOp predictBatchOp = new IsotonicRegPredictBatchOp().setPredictionCol("predictCol"); predictBatchOp.linkFrom(model,predictData).print(); }