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); } ClusterEvaluationUtil(ContinuousDistance distance); static Params extractParamsFromConfusionMatrix(LongMatrix longMatrix); static Tuple1<Double> calSilhouetteCoefficient(Row row, ClusterMetricsSummary clusterMetricsSummary); static Params getBasicClusterStatistics(Iterable<Row> rows); static ClusterMetricsSummary getClusterStatistics(Iterable<Row> rows, ContinuousDistance distance); @Override void apply(TimeWindow timeWindow, Iterable<Row> rows, Collector<Row> collector); static int COUNT; static int MEAN; static int SUM_2; }
@Test public void getClusterStatisticsEuclideanTest() { Row[] rows0 = new Row[] { Row.of(0, "0,0,0"), Row.of(0, "0.1,0.1,0.1"), Row.of(0, "0.2,0.2,0.2") }; ClusterMetricsSummary clusterMetricsSummary = ClusterEvaluationUtil.getClusterStatistics(Arrays.asList(rows0), new EuclideanDistance()); Assert.assertEquals(clusterMetricsSummary.k, 1); Assert.assertEquals(clusterMetricsSummary.clusterId.get(0), "0"); Assert.assertEquals(clusterMetricsSummary.clusterCnt.get(0).intValue(), 3); Assert.assertEquals(clusterMetricsSummary.compactness.get(0), 0.115, 0.001); Assert.assertEquals(clusterMetricsSummary.distanceSquareSum.get(0), 0.06, 0.01); Assert.assertEquals(clusterMetricsSummary.vectorNormL2Sum.get(0), 0.15, 0.01); Assert.assertEquals(clusterMetricsSummary.meanVector.get(0), new DenseVector(new double[]{0.1, 0.1, 0.1})); Assert.assertEquals(clusterMetricsSummary.k, 1); Assert.assertEquals(clusterMetricsSummary.total, 3); }
@Test public void getClusterStatisticsCosineTest() { Row[] rows0 = new Row[] { Row.of(1, "9 9 9"), Row.of(1, "9.1 9.1 9.1"), Row.of(1, "9.2 9.2 9.2") }; ClusterMetricsSummary clusterMetricsSummary = ClusterEvaluationUtil.getClusterStatistics(Arrays.asList(rows0), new CosineDistance()); Assert.assertEquals(clusterMetricsSummary.k, 1); Assert.assertEquals(clusterMetricsSummary.clusterId.get(0), "1"); Assert.assertEquals(clusterMetricsSummary.clusterCnt.get(0).intValue(), 3); Assert.assertEquals(clusterMetricsSummary.compactness.get(0), 0, 0.001); Assert.assertEquals(clusterMetricsSummary.distanceSquareSum.get(0), 0, 0.01); Assert.assertEquals(clusterMetricsSummary.vectorNormL2Sum.get(0), 3.0, 0.01); Assert.assertEquals(clusterMetricsSummary.meanVector.get(0).normL2Square(), 1.0, 0.01); Assert.assertEquals(clusterMetricsSummary.k, 1); Assert.assertEquals(clusterMetricsSummary.total, 3); }