class EvalClusterBatchOp extends BatchOperator<EvalClusterBatchOp> implements EvalClusterParams<EvalClusterBatchOp>, EvaluationMetricsCollector<ClusterMetrics, EvalClusterBatchOp> {

    @Override
    public EvalClusterBatchOp linkFrom(BatchOperator<?>... inputs) {
        BatchOperator in = checkAndGetFirst(inputs);
        String labelColName = this.getLabelCol();
        String predResultColName = this.getPredictionCol();
        String vectorColName = this.getVectorCol();
        DistanceType distanceType = getDistanceType();
        FastDistance distance = distanceType.getFastDistance();

        DataSet<Params> empty = MLEnvironmentFactory.get(getMLEnvironmentId()).getExecutionEnvironment().fromElements(
            new Params());
        DataSet<Params> labelMetrics = empty, vectorMetrics;

        if (null != labelColName) {
            DataSet<Row> data = in.select(new String[] {labelColName, predResultColName}).getDataSet();
            labelMetrics = calLocalPredResult(data)
                .reduce(new ReduceFunction<LongMatrix>() {
                    @Override
                    public LongMatrix reduce(LongMatrix value1, LongMatrix value2) {
                        value1.plusEqual(value2);
                        return value1;
                    }
                })
                .map(new MapFunction<LongMatrix, Params>() {
                    @Override
                    public Params map(LongMatrix value) {
                        return ClusterEvaluationUtil.extractParamsFromConfusionMatrix(value);
                    }
                });
        }
        if (null != vectorColName) {
            DataSet<Row> data = in.select(new String[] {predResultColName, vectorColName}).getDataSet();
            DataSet<BaseMetricsSummary> metricsSummary = data
                .groupBy(0)
                .reduceGroup(new CalcClusterMetricsSummary(distance))
                .reduce(new EvaluationUtil.ReduceBaseMetrics());
            DataSet<Tuple1<Double>> silhouetteCoefficient = data.map(
                new RichMapFunction<Row, Tuple1<Double>>() {
                    @Override
                    public Tuple1<Double> map(Row value) {
                        List<BaseMetricsSummary> list = getRuntimeContext().getBroadcastVariable(METRICS_SUMMARY);
                        return ClusterEvaluationUtil.calSilhouetteCoefficient(value,
                            (ClusterMetricsSummary)list.get(0));
                    }
                }).withBroadcastSet(metricsSummary, METRICS_SUMMARY)
                .aggregate(Aggregations.SUM, 0);

            vectorMetrics = metricsSummary.map(new ClusterEvaluationUtil.SaveDataAsParams()).withBroadcastSet(
                silhouetteCoefficient, SILHOUETTE_COEFFICIENT);
        } else {
            vectorMetrics = in.select(predResultColName)
                .getDataSet()
                .reduceGroup(new BasicClusterParams());
        }

        DataSet<Row> out = labelMetrics
            .union(vectorMetrics)
            .reduceGroup(new GroupReduceFunction<Params, Row>() {
                @Override
                public void reduce(Iterable<Params> values, Collector<Row> out) {
                    Params params = new Params();
                    for (Params p : values) {
                        params.merge(p);
                    }
                    out.collect(Row.of(params.toJson()));
                }
            });

        this.setOutputTable(DataSetConversionUtil.toTable(getMLEnvironmentId(),
            out, new TableSchema(new String[] {EVAL_RESULT}, new TypeInformation[] {Types.STRING})
        ));
        return this;
    }

    public  EvalClusterBatchOp();
    public  EvalClusterBatchOp(Params params);

}