class 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();
        // initialize the selectedColNames.
        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!");
        }
        //initialize the input data, the three dimensions are label, feature, weight.
        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;
    }

    public  IsotonicRegTrainBatchOp();
    public  IsotonicRegTrainBatchOp(Params params);

}