ClusterEvaluationUtil implements AllWindowFunction<Row, Row, TimeWindow> { public static Params getBasicClusterStatistics(Iterable<Row> rows){ Map<String, Double> map = new HashMap<>(0); int count = 0; for (Row row : rows) { if(row != null && row.getField(0) != null) { count++; String key = row.getField(0).toString(); map.merge(key, 1.0, (k,v) -> k + 1.0); } } int c = 0; double[] values = new double[map.size()]; String[] keys = new String[map.size()]; for(Map.Entry<String, Double> entry : map.entrySet()){ keys[c] = entry.getKey(); values[c++] = entry.getValue(); } return new Params() .set(ClusterMetrics.COUNT, count) .set(ClusterMetrics.K, map.size()) .set(ClusterMetrics.CLUSTER_ARRAY, keys) .set(ClusterMetrics.COUNT_ARRAY, values); } 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 getBasicClusterStatisticsTest() { Row[] rows = new Row[] { Row.of(0), Row.of(0), Row.of(0), Row.of(1), Row.of(1) }; Params params = ClusterEvaluationUtil.getBasicClusterStatistics(Arrays.asList(rows)); Assert.assertEquals(params.get(ClusterMetrics.COUNT).intValue(), 5); Assert.assertEquals(params.get(ClusterMetrics.K).intValue(), 2); Assert.assertArrayEquals(params.get(ClusterMetrics.CLUSTER_ARRAY), new String[]{"0", "1"}); }