class ClusterEvaluationUtil implements AllWindowFunction<Row, Row, TimeWindow> {

    public static ClusterMetricsSummary getClusterStatistics(Iterable<Row> rows, ContinuousDistance distance) {
        List<Vector> list = new ArrayList<>();
        int total = 0;
        String clusterId;
        DenseVector sumVector;

        Iterator<Row> iterator = rows.iterator();
        Row row = null;
        while (iterator.hasNext() && !EvaluationUtil.checkRowFieldNotNull(row)) {
            row = iterator.next();
        }
        if (EvaluationUtil.checkRowFieldNotNull(row)) {
            clusterId = row.getField(0).toString();
            Vector vec = VectorUtil.getVector(row.getField(1));
            Preconditions.checkArgument(vec.size() >= 1, "Vector Size must be at least 1!");
            sumVector = DenseVector.zeros(vec.size());
        } else {
            return null;
        }

        while (null != row) {
            if (EvaluationUtil.checkRowFieldNotNull(row)) {
                Preconditions.checkArgument(row.getField(0).toString().equals(clusterId),
                    "ClusterId must be the same!");
                Vector vec = VectorUtil.getVector(row.getField(1));
                list.add(vec);
                if (distance instanceof EuclideanDistance) {
                    sumVector.plusEqual(vec);
                } else {
                    vec.scaleEqual(1.0 / vec.normL2());
                    sumVector.plusEqual(vec);
                }
                total++;
            }
            row = iterator.hasNext() ? iterator.next() : null;
        }

        DenseVector meanVector = sumVector.scale(1.0 / total);

        double distanceSum = 0.0;
        double distanceSquareSum = 0.0;
        double vectorNormL2Sum = 0.0;
        for (Vector vec : list) {
            double d = distance.calc(meanVector, vec);
            distanceSum += d;
            distanceSquareSum += d * d;
            vectorNormL2Sum += vec.normL2Square();
        }
        return new ClusterMetricsSummary(clusterId, total, distanceSum / total, distanceSquareSum, vectorNormL2Sum,
            meanVector, distance);
    }

    public  ClusterEvaluationUtil(ContinuousDistance distance);

}