Repository: skiptests/astraea
Pull request #382 (2022-05-28 07:40:30)

Move all entropy code out of NeutralIntegratedCost
這隻PR主要的目的是將NeutralIntegratedCost計算權重的部分抽象畫出來，如此才不會有一堆數學計算擠在一起，也方便之後我們針對不同階段的運算做優化，這次抽出來的項目有兩個：
1. normalization，並提供預設的min-max normalization
2. weight provider，也提供了預設基於entropy的實作

chia7712: 目前測試會失敗，但因為有些細節還需要討論，之後會修正
chia7712: @chinghongfang @wycccccc 上面有幾個討論是延續上禮拜三的故事，再麻煩你們幫我解惑一下，感恩

Patch:
diff --git a/app/src/main/java/org/astraea/cost/Normalizer.java b/app/src/main/java/org/astraea/cost/Normalizer.java
new file mode 100644
index 00000000..8dba2999
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/Normalizer.java
@@ -0,0 +1,64 @@
+package org.astraea.cost;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/** used to normalize data into a range between [0, 1] */
+public interface Normalizer {
+
+  /** @return all normalizers */
+  static List<Normalizer> all() {
+    return List.of(Normalizer.proportion(), Normalizer.minMax(true), Normalizer.minMax(false));
+  }
+
+  /**
+   * implement the min-max normalization.
+   *
+   * <p>positive indexes: (value - min) / (max - min) negative
+   *
+   * <p>indexes: (max - value) / (max - min)
+   *
+   * @param positive true if the data is positive indexes. Otherwise, false
+   * @return min-max normalizer
+   */
+  static Normalizer minMax(boolean positive) {
+    var comparator = Comparator.comparing(Double::doubleValue);
+    return values -> {
+      double max = values.stream().max(comparator).orElse(0.0);
+      double min = values.stream().min(comparator).orElse(0.0);
+      // there is nothing to rescale, so we just all same values
+      if (max == min)
+        return IntStream.range(0, values.size())
+            .mapToObj(ignored -> 1.0)
+            .collect(Collectors.toUnmodifiableList());
+      return values.stream()
+          .map(value -> (positive ? value - min : max - value) / (max - min))
+          .collect(Collectors.toUnmodifiableList());
+    };
+  }
+
+  /**
+   * rescale the value by the proportion
+   *
+   * <p>value / sum of all values
+   *
+   * @return proportion normalizer
+   */
+  static Normalizer proportion() {
+    return values -> {
+      var sum = values.stream().mapToDouble(i -> i).sum();
+      return values.stream().map(v -> v / sum).collect(Collectors.toUnmodifiableList());
+    };
+  }
+
+  /**
+   * rescales the values into a range of [0,1]
+   *
+   * @param values origin data
+   * @return rescaled data
+   */
+  Collection<Double> normalize(Collection<Double> values);
+}
diff --git a/app/src/main/java/org/astraea/cost/WeightProvider.java b/app/src/main/java/org/astraea/cost/WeightProvider.java
new file mode 100644
index 00000000..0cb9b66e
--- /dev/null
+++ b/app/src/main/java/org/astraea/cost/WeightProvider.java
@@ -0,0 +1,65 @@
+package org.astraea.cost;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** Used to provide the weight to score the node or partition */
+@FunctionalInterface
+public interface WeightProvider {
+
+  /**
+   * create a weight provider based on entropy
+   *
+   * @param normalizer to normalize input values
+   * @return weight for each metrics
+   */
+  static WeightProvider entropy(Normalizer normalizer) {
+    return new EntropyWeightProvider(normalizer);
+  }
+
+  /**
+   * compute the weights for each metric
+   *
+   * @param <Metrics> used to represent the "resource". For example, throughput, memory usage, etc.
+   * @param <T> used to represent the "target". For example, broker node, topic partition, etc.
+   * @param values origin data
+   * @return metric and its weight
+   */
+  <Metrics, T extends Collection<Double>> Map<Metrics, Double> weight(Map<Metrics, T> values);
+
+  class EntropyWeightProvider implements WeightProvider {
+    private final Normalizer normalizer;
+
+    EntropyWeightProvider(Normalizer normalizer) {
+      this.normalizer = normalizer;
+    }
+
+    @Override
+    public <Metrics, T extends Collection<Double>> Map<Metrics, Double> weight(
+        Map<Metrics, T> values) {
+      var entropies = entropies(values);
+      // use difference to calculate the weight
+      var diff =
+          entropies.entrySet().stream()
+              .collect(Collectors.toMap(Map.Entry::getKey, e -> 1D - e.getValue()));
+      var sum = diff.values().stream().mapToDouble(d -> d).sum();
+      return diff.entrySet().stream()
+          .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() / sum));
+    }
+
+    // used for testing
+    <Metric, T extends Collection<Double>> Map<Metric, Double> entropies(Map<Metric, T> values) {
+      return values.entrySet().stream()
+          .collect(
+              Collectors.toMap(
+                  Map.Entry::getKey,
+                  e ->
+                      normalizer.normalize(e.getValue()).stream()
+                              // return 0 if value is 0 (just convenience of calculation)
+                              .mapToDouble(value -> value * (value == 0 ? 0 : Math.log(value)))
+                              .sum()
+                          / (-Math.log(e.getValue().size()))));
+    }
+  }
+}
diff --git a/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java b/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
index 10ebba84..50316d21 100644
--- a/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
+++ b/app/src/main/java/org/astraea/cost/broker/NeutralIntegratedCost.java
@@ -1,6 +1,6 @@
 package org.astraea.cost.broker;
 
-import java.util.Comparator;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -9,6 +9,8 @@ import org.astraea.cost.BrokerCost;
 import org.astraea.cost.ClusterInfo;
 import org.astraea.cost.CostFunction;
 import org.astraea.cost.HasBrokerCost;
+import org.astraea.cost.Normalizer;
+import org.astraea.cost.WeightProvider;
 import org.astraea.metrics.collector.Fetcher;
 
 /**
@@ -22,8 +24,8 @@ public class NeutralIntegratedCost implements HasBrokerCost {
   private final List<HasBrokerCost> metricsCost =
       List.of(new BrokerInputCost(), new BrokerOutputCost(), new CpuCost(), new MemoryCost());
   private final Map<Integer, BrokerMetrics> brokersMetric = new HashMap<>();
-  private final EntropyEmpowerment entropyEmpowerment = new EntropyEmpowerment();
   private final AHPEmpowerment ahpEmpowerment = new AHPEmpowerment();
+  private final WeightProvider weightProvider = WeightProvider.entropy(Normalizer.minMax(true));
 
   @Override
   public BrokerCost brokerCost(ClusterInfo clusterInfo) {
@@ -63,7 +65,7 @@ public class NeutralIntegratedCost implements HasBrokerCost {
         });
 
     var integratedEmpowerment =
-        entropyEmpowerment.empowerment(brokersMetric).entrySet().stream()
+        weight(weightProvider, brokersMetric).entrySet().stream()
             .collect(
                 Collectors.toMap(
                     Map.Entry::getKey,
@@ -87,6 +89,39 @@ public class NeutralIntegratedCost implements HasBrokerCost {
                                 * integratedEmpowerment.get(Metrics.memory.metricName)));
   }
 
+  static Map<String, Double> weight(
+      WeightProvider weightProvider, Map<Integer, BrokerMetrics> brokerMetrics) {
+
+    var values =
+        Arrays.stream(Metrics.values())
+            .collect(
+                Collectors.toMap(
+                    m -> m.metricName,
+                    m -> {
+                      switch (m) {
+                        case inputThroughput:
+                          return brokerMetrics.values().stream()
+                              .map(metrics -> metrics.inputScore)
+                              .collect(Collectors.toUnmodifiableList());
+                        case outputThroughput:
+                          return brokerMetrics.values().stream()
+                              .map(metrics -> metrics.outputScore)
+                              .collect(Collectors.toUnmodifiableList());
+                        case memory:
+                          return brokerMetrics.values().stream()
+                              .map(metrics -> metrics.memoryScore)
+                              .collect(Collectors.toUnmodifiableList());
+                        case cpu:
+                          return brokerMetrics.values().stream()
+                              .map(metrics -> metrics.cpuScore)
+                              .collect(Collectors.toUnmodifiableList());
+                        default:
+                          return List.<Double>of();
+                      }
+                    }));
+    return weightProvider.weight(values);
+  }
+
   @Override
   public Fetcher fetcher() {
     return Fetcher.of(metricsCost.stream().map(CostFunction::fetcher).collect(Collectors.toList()));
@@ -101,92 +136,6 @@ public class NeutralIntegratedCost implements HasBrokerCost {
     BrokerMetrics() {}
   }
 
-  /**
-   * Entropy weight method (EWM) is a commonly used weighting method that measures value dispersion
-   * in decision-making. The greater the degree of dispersion, the greater the degree of
-   * differentiation, and more information can be derived. Meanwhile, higher weight should be given
-   * to the index, and vice versa.
-   */
-  static class EntropyEmpowerment {
-    public Map<String, Double> empowerment(Map<Integer, BrokerMetrics> brokerMetrics) {
-      var metrics = new HashMap<String, Double>();
-      metrics.put(
-          Metrics.inputThroughput.metricName,
-          entropy(
-              brokerMetrics.entrySet().stream()
-                  .collect(
-                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().inputScore))));
-      metrics.put(
-          Metrics.outputThroughput.metricName,
-          entropy(
-              brokerMetrics.entrySet().stream()
-                  .collect(
-                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().outputScore))));
-      metrics.put(
-          Metrics.memory.metricName,
-          entropy(
-              brokerMetrics.entrySet().stream()
-                  .collect(
-                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().memoryScore))));
-      metrics.put(
-          Metrics.cpu.metricName,
-          entropy(
-              brokerMetrics.entrySet().stream()
-                  .collect(
-                      Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().cpuScore))));
-      var sum = metrics.values().stream().mapToDouble(i -> i).sum();
-
-      return metrics.entrySet().stream()
-          .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue() / sum));
-    }
-
-    /**
-     * To the cost indexes
-     *
-     * @param metrics JMX metrics for each broker.
-     * @return Standardization of result
-     */
-    Map<Integer, Double> negativeIndicator(Map<Integer, Double> metrics) {
-      Comparator<Double> comparator = Comparator.comparing(Double::doubleValue);
-      var max = metrics.values().stream().max(comparator).orElse(0.0);
-      var min = metrics.values().stream().min(comparator).orElse(0.0);
-      return metrics.entrySet().stream()
-          .collect(
-              Collectors.toMap(Map.Entry::getKey, entry -> (max - entry.getValue()) / (max - min)));
-    }
-
-    /**
-     * To the benefit indexes.
-     *
-     * @param metrics JMX metrics for each broker.
-     * @return Standardization of result
-     */
-    Map<Integer, Double> positiveIndicator(Map<Integer, Double> metrics) {
-      Comparator<Double> comparator = Comparator.comparing(Double::doubleValue);
-      var max = metrics.values().stream().max(comparator).orElse(0.0) + 0.01;
-      double min = metrics.values().stream().min(comparator).orElse(0.0);
-      return metrics.entrySet().stream()
-          .collect(
-              Collectors.toMap(
-                  Map.Entry::getKey, entry -> (entry.getValue() + 0.01 - min) / (max - min)));
-    }
-
-    Map<Integer, Double> metricsWeight(Map<Integer, Double> indicator) {
-      var sum = indicator.values().stream().mapToDouble(i -> i).sum();
-      return indicator.entrySet().stream()
-          .collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue() / sum));
-    }
-
-    double entropy(Map<Integer, Double> metrics) {
-      return 1
-          - metricsWeight(positiveIndicator(metrics)).values().stream()
-                  .filter(weight -> !weight.equals(0.0))
-                  .mapToDouble(weight -> weight * Math.log(weight))
-                  .sum()
-              / (-Math.log(metrics.size()));
-    }
-  }
-
   /**
    * AHP is a structured technique for organizing and solving complex decision-making problems based
    * on mathematics and psychology. AHP provides a comprehensive and logical framework to quantify
@@ -236,8 +185,4 @@ public class NeutralIntegratedCost implements HasBrokerCost {
       return metricName;
     }
   }
-
-  EntropyEmpowerment entropyEmpowerment() {
-    return entropyEmpowerment;
-  }
 }


Test patch:
diff --git a/app/src/test/java/org/astraea/cost/NormalizerTest.java b/app/src/test/java/org/astraea/cost/NormalizerTest.java
new file mode 100644
index 00000000..cefa960f
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/NormalizerTest.java
@@ -0,0 +1,57 @@
+package org.astraea.cost;
+
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class NormalizerTest {
+
+  @ParameterizedTest
+  @MethodSource("normalizers")
+  void testRandomValues(Normalizer normalizer) {
+    IntStream.range(0, 100)
+        .forEach(
+            index -> {
+              // generate random data
+              var data =
+                  IntStream.range(0, 100)
+                      .boxed()
+                      .map(i -> Math.random() * i * 10000)
+                      .collect(Collectors.toUnmodifiableList());
+              var result = normalizer.normalize(data);
+              Assertions.assertNotEquals(0, result.size());
+              // make sure there is no NaN
+              result.forEach(v -> Assertions.assertNotEquals(Double.NaN, v));
+              // make sure all values are in the range between 0 and 1
+              result.forEach(v -> Assertions.assertTrue(0 <= v && v <= 1));
+            });
+  }
+
+  @ParameterizedTest
+  @MethodSource("normalizers")
+  void testSmallValues(Normalizer normalizer) {
+    IntStream.range(0, 100)
+        .forEach(
+            index -> {
+              // generate random data
+              var data =
+                  IntStream.range(0, 100)
+                      .mapToObj(i -> Math.max(0.3, Math.min(0.7, Math.random())))
+                      .collect(Collectors.toUnmodifiableList());
+              var result = normalizer.normalize(data);
+              Assertions.assertNotEquals(0, result.size());
+              // make sure there is no NaN
+              result.forEach(v -> Assertions.assertNotEquals(Double.NaN, v));
+              // make sure all values are in the range between 0 and 1
+              result.forEach(v -> Assertions.assertTrue(0 <= v && v <= 1));
+            });
+  }
+
+  private static Stream<Arguments> normalizers() {
+    return Normalizer.all().stream().map(Arguments::of);
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/WeightProviderTest.java b/app/src/test/java/org/astraea/cost/WeightProviderTest.java
new file mode 100644
index 00000000..c72080b5
--- /dev/null
+++ b/app/src/test/java/org/astraea/cost/WeightProviderTest.java
@@ -0,0 +1,53 @@
+package org.astraea.cost;
+
+import java.util.Comparator;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+public class WeightProviderTest {
+
+  @ParameterizedTest
+  @MethodSource("normalizers")
+  void testEntropy(Normalizer normalizer) {
+    var length = 8;
+    var numberOfMetrics = 2;
+    var numberOfObjects = 2;
+    var weightProvider = new WeightProvider.EntropyWeightProvider(normalizer);
+    var raw =
+        IntStream.range(0, numberOfMetrics)
+            .boxed()
+            .collect(
+                Collectors.toMap(
+                    String::valueOf,
+                    ignored ->
+                        IntStream.range(0, numberOfObjects)
+                            .mapToObj(i -> Math.random())
+                            .collect(Collectors.toUnmodifiableList())));
+    // The smaller entropy means larger weight, so we sort the entropies/weights by different order
+    // to check the metrics name later
+    var entropies =
+        weightProvider.entropies(raw).entrySet().stream()
+            .sorted(Map.Entry.comparingByValue())
+            .collect(Collectors.toUnmodifiableList());
+
+    entropies.forEach(e -> Assertions.assertTrue(0 <= e.getValue() && e.getValue() <= 1));
+
+    var weights =
+        weightProvider.weight(raw).entrySet().stream()
+            .sorted(Map.Entry.comparingByValue(Comparator.reverseOrder()))
+            .collect(Collectors.toUnmodifiableList());
+    Assertions.assertEquals(entropies.size(), weights.size());
+    IntStream.range(0, entropies.size())
+        .forEach(i -> Assertions.assertEquals(entropies.get(i).getKey(), weights.get(i).getKey()));
+  }
+
+  private static Stream<Arguments> normalizers() {
+    return Normalizer.all().stream().map(Arguments::of);
+  }
+}
diff --git a/app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java b/app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java
deleted file mode 100644
index 8145ed23..00000000
--- a/app/src/test/java/org/astraea/cost/broker/NeutralIntegratedCostTest.java
+++ /dev/null
@@ -1,42 +0,0 @@
-package org.astraea.cost.broker;
-
-import java.util.Map;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-// TODO Add more test.
-public class NeutralIntegratedCostTest {
-  @Test
-  void testEntropyEmpowerment() {
-    var neutralIntegratedCost = new NeutralIntegratedCost();
-    var brokers =
-        Map.of(
-            0,
-            new NeutralIntegratedCost.BrokerMetrics(),
-            1,
-            new NeutralIntegratedCost.BrokerMetrics(),
-            2,
-            new NeutralIntegratedCost.BrokerMetrics());
-    brokers.get(0).cpuScore = 0.1252;
-    brokers.get(0).inputScore = 550.0;
-    brokers.get(0).outputScore = 545.0;
-    brokers.get(0).memoryScore = 0.254;
-
-    brokers.get(1).inputScore = 750.0;
-    brokers.get(1).outputScore = 755.0;
-    brokers.get(1).memoryScore = 0.268;
-    brokers.get(1).cpuScore = 0.1342;
-
-    brokers.get(2).inputScore = 550.0;
-    brokers.get(2).outputScore = 550.0;
-    brokers.get(2).memoryScore = 0.235;
-    brokers.get(2).cpuScore = 0.1522;
-
-    var metrics = neutralIntegratedCost.entropyEmpowerment().empowerment(brokers);
-
-    Assertions.assertEquals(0.46689278143812796, metrics.get("inputThroughput"));
-    Assertions.assertEquals(0.4200836938888968, metrics.get("outputThroughput"));
-    Assertions.assertEquals(0.05781512136314425, metrics.get("memory"));
-    Assertions.assertEquals(0.055208403309831, metrics.get("cpu"));
-  }
-}