AmbariMetricsServiceWithStormQuerier extends AbstractTimeSeriesQuerier { @VisibleForTesting Map<Long, Double> aggregateWithApplyingWeightedAverage(Map<Long, List<Pair<String, Double>>> keyMetric, Map<Long, List<Pair<String, Double>>> weightMetric) { Map<Long, Double> ret = new HashMap<>(); for (Map.Entry<Long, List<Pair<String, Double>>> keyMetricEntry : keyMetric.entrySet()) { long timestamp = keyMetricEntry.getKey(); List<Pair<String, Double>> keyStreamToValueList = keyMetricEntry.getValue(); List<Pair<String, Double>> weightStreamToValueList = weightMetric.get(timestamp); if (weightStreamToValueList == null || weightStreamToValueList.isEmpty()) { ret.put(timestamp, 0.0d); continue; } Double totalWeight = weightStreamToValueList.stream().mapToDouble(p -> p.getRight()).sum(); if (DoubleUtils.equalsToZero(totalWeight)) { ret.put(timestamp, 0.0d); continue; } double weightedSum = keyStreamToValueList.stream().map(pair -> { String stream = pair.getLeft(); Double value = pair.getRight(); Double weightForStream = weightStreamToValueList.stream() .filter(p -> p.getLeft().equals(stream)).findAny().map(op -> op.getRight()).orElse(0.0); Double weight = weightForStream / totalWeight; return value * weight; }).mapToDouble(d -> d).sum(); ret.put(timestamp, weightedSum); } return ret; } AmbariMetricsServiceWithStormQuerier(); @Override void init(Map<String, String> conf); @Override Map<Long, Double> getTopologyLevelMetrics(String topologyName, String metricName,
                                                     AggregateFunction aggrFunction, long from, long to); @Override Map<Long, Double> getMetrics(String topologyName, String componentId, String metricName,
                                        AggregateFunction aggrFunction, long from, long to); @Override Map<String, Map<Long, Double>> getRawMetrics(String metricName, String parameters, long from, long to); static final String METRIC_NAME_PREFIX_KAFKA_OFFSET; }
@Test public void testAggregateWithWeightedAverage() { Map<Long, List<Pair<String, Double>>> keyMetric = new HashMap<>(); List<Pair<String, Double>> keyPoints1 = new ArrayList<>(); keyPoints1.add(Pair.of("stream1", 10.0)); keyPoints1.add(Pair.of("stream2", 20.0)); keyMetric.put(1L, keyPoints1); List<Pair<String, Double>> keyPoints2 = new ArrayList<>(); keyPoints2.add(Pair.of("stream1", 10.0)); keyPoints2.add(Pair.of("stream2", 20.0)); keyMetric.put(2L, keyPoints2); Map<Long, List<Pair<String, Double>>> weightMetric = new HashMap<>(); List<Pair<String, Double>> weightPoints1 = new ArrayList<>(); weightPoints1.add(Pair.of("stream1", 10.0)); weightPoints1.add(Pair.of("stream2", 5.0)); weightMetric.put(1L, weightPoints1); List<Pair<String, Double>> weightPoints2 = new ArrayList<>(); weightPoints2.add(Pair.of("stream1", 5.0)); weightPoints2.add(Pair.of("stream2", 10.0)); weightMetric.put(2L, weightPoints2); Map<Long, Double> ret = querier.aggregateWithApplyingWeightedAverage(keyMetric, weightMetric); Assert.assertEquals(2, ret.size()); Double aggregated = ret.get(1L); Assert.assertNotNull(aggregated); Double expected = 10.0 * 10.0 / (10.0 + 5.0) + 20.0 * 5.0 / (10.0 + 5.0); Assert.assertEquals(expected, aggregated, 0.00001d); aggregated = ret.get(2L); Assert.assertNotNull(aggregated); expected = 10.0 * 5.0 / (5.0 + 10.0) + 20.0 * 10.0 / (5.0 + 10.0); Assert.assertEquals(expected, aggregated, 0.00001d); }
@Test public void testAggregateWithWeightedAverageLacksWeightInformation() { Map<Long, List<Pair<String, Double>>> keyMetric = new HashMap<>(); List<Pair<String, Double>> keyPoints1 = new ArrayList<>(); keyPoints1.add(Pair.of("stream1", 10.0)); keyPoints1.add(Pair.of("stream2", 20.0)); keyMetric.put(1L, keyPoints1); List<Pair<String, Double>> keyPoints2 = new ArrayList<>(); keyPoints2.add(Pair.of("stream1", 10.0)); keyPoints2.add(Pair.of("stream2", 20.0)); keyMetric.put(2L, keyPoints2); List<Pair<String, Double>> keyPoints3 = new ArrayList<>(); keyPoints3.add(Pair.of("stream1", 10.0)); keyPoints3.add(Pair.of("stream2", 20.0)); keyMetric.put(3L, keyPoints3); Map<Long, List<Pair<String, Double>>> weightMetric = new HashMap<>(); List<Pair<String, Double>> weightPoints2 = new ArrayList<>(); weightPoints2.add(Pair.of("stream1", 0.0)); weightPoints2.add(Pair.of("stream2", 0.0)); weightMetric.put(2L, weightPoints2); List<Pair<String, Double>> weightPoints3 = new ArrayList<>(); weightPoints3.add(Pair.of("stream1", 10.0)); weightMetric.put(3L, weightPoints3); Map<Long, Double> ret = querier.aggregateWithApplyingWeightedAverage(keyMetric, weightMetric); Assert.assertEquals(3, ret.size()); Double aggregated = ret.get(1L); Assert.assertNotNull(aggregated); Assert.assertEquals(0.0d, aggregated, 0.00001d); aggregated = ret.get(2L); Assert.assertNotNull(aggregated); Assert.assertEquals(0.0d, aggregated, 0.00001d); aggregated = ret.get(3L); Assert.assertNotNull(aggregated); Assert.assertEquals(10.0d, aggregated, 0.00001d); }