Added AveragePairWithStats and parallelGroupedNumberPairDataAverageAggregationProcessor

java/com.sap.sse.datamining.shared/src/com/sap/sse/datamining/shared/data/AveragePairWithStats.java

@@ -0,0 +1,29 @@
+package com.sap.sse.datamining.shared.data;
+
+import java.io.Serializable;
+
+import com.sap.sse.common.Util.Pair;
+
+/**
+ * Result of an averaging operation, augmented by statistical data explaining a bit more about how the average was
+ * computed, including minimum and maximum values, count, median, as well as standard deviation. All but count will have
+ * the same type of the average value whereas the count is an integer-like data type.
+ * 
+ * @author Axel Uhl (d043530)
+ *
+ * @param <T>
+ */
+public interface AveragePairWithStats<T> extends Serializable {
+    Pair<T, T> getAverage();
+    Pair<T, T> getMin();
+    Pair<T, T> getMax();
+    Pair<T, T> getMedian();
+    Pair<T, T> getStandardDeviation();
+    long getCount();
+    
+    /**
+     * @return the fully-qualified type name of the {@code T} type argument; may be used by
+     * the results presenter to choose an applicable data provider.
+     */
+    String getResultType();
+}

java/com.sap.sse.datamining.shared/src/com/sap/sse/datamining/shared/impl/AveragePairWithStatsImpl.java

@@ -0,0 +1,125 @@
+package com.sap.sse.datamining.shared.impl;
+
+import com.sap.sse.common.Util.Pair;
+import com.sap.sse.datamining.shared.data.AveragePairWithStats;
+
+public class AveragePairWithStatsImpl<T> implements AveragePairWithStats<T> {
+    private static final long serialVersionUID = -3094183173270210846L;
+    private final Pair<T, T> average;
+    private final Pair<T, T> min;
+    private final Pair<T, T> max;
+    private final Pair<T, T> median;
+    private final Pair<T, T> standardDeviation;
+    private final long count;
+    private final String resultType;
+
+    public AveragePairWithStatsImpl(Pair<T, T> average, Pair<T, T> min, Pair<T, T> max, Pair<T, T> median, Pair<T, T> standardDeviation, long count, String resultType) {
+        super();
+        this.average = average;
+        this.min = min;
+        this.max = max;
+        this.median = median;
+        this.standardDeviation = standardDeviation;
+        this.count = count;
+        this.resultType = resultType;
+    }
+
+    @Override
+    public Pair<T, T> getAverage() {
+        return average;
+    }
+
+    @Override
+    public Pair<T, T> getMin() {
+        return min;
+    }
+
+    @Override
+    public Pair<T, T> getMax() {
+        return max;
+    }
+
+    @Override
+    public Pair<T, T> getMedian() {
+        return median;
+    }
+
+    @Override
+    public Pair<T, T> getStandardDeviation() {
+        return standardDeviation;
+    }
+
+    @Override
+    public long getCount() {
+        return count;
+    }
+
+    @Override
+    public String getResultType() {
+        return resultType;
+    }
+
+    @Override
+    public String toString() {
+        return "AverageWithStatsImpl [average=" + average + ", min=" + min + ", max=" + max + ", median=" + median
+                + ", standardDeviation=" + standardDeviation + ", count=" + count + ", resultType=" + resultType + "]";
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((average == null) ? 0 : average.hashCode());
+        result = prime * result + (int) (count ^ (count >>> 32));
+        result = prime * result + ((max == null) ? 0 : max.hashCode());
+        result = prime * result + ((median == null) ? 0 : median.hashCode());
+        result = prime * result + ((min == null) ? 0 : min.hashCode());
+        result = prime * result + ((resultType == null) ? 0 : resultType.hashCode());
+        result = prime * result + ((standardDeviation == null) ? 0 : standardDeviation.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        AveragePairWithStatsImpl<?> other = (AveragePairWithStatsImpl<?>) obj;
+        if (average == null) {
+            if (other.average != null)
+                return false;
+        } else if (!average.equals(other.average))
+            return false;
+        if (count != other.count)
+            return false;
+        if (max == null) {
+            if (other.max != null)
+                return false;
+        } else if (!max.equals(other.max))
+            return false;
+        if (median == null) {
+            if (other.median != null)
+                return false;
+        } else if (!median.equals(other.median))
+            return false;
+        if (min == null) {
+            if (other.min != null)
+                return false;
+        } else if (!min.equals(other.min))
+            return false;
+        if (resultType == null) {
+            if (other.resultType != null)
+                return false;
+        } else if (!resultType.equals(other.resultType))
+            return false;
+        if (standardDeviation == null) {
+            if (other.standardDeviation != null)
+                return false;
+        } else if (!standardDeviation.equals(other.standardDeviation))
+            return false;
+        return true;
+    }
+}

java/com.sap.sse.datamining/resources/stringmessages/StringMessages.properties

@@ -1,4 +1,5 @@
 Average=Average
+PairAverage=Average (Pair)
 Median=Median
 Sum=Sum
 Maximum=Maximum

java/com.sap.sse.datamining/resources/stringmessages/StringMessages_de.properties

@@ -1,4 +1,5 @@
 Average=Durchschnitt
+PairAverage=Durchschnitt (Paar)
 Median=Median
 Sum=Summe
 Maximum=Maximum

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/DataMiningFrameworkActivator.java

@@ -33,6 +33,7 @@ import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberD
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMedianAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMinAggregationProcessor;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataSumAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberPairDataAverageAggregationProcessor;
 import com.sap.sse.datamining.impl.components.management.AggregationProcessorDefinitionManager;
 import com.sap.sse.datamining.impl.components.management.DataRetrieverChainDefinitionManager;
 import com.sap.sse.datamining.impl.components.management.DataSourceProviderManager;
@@ -86,6 +87,7 @@ public class DataMiningFrameworkActivator implements BundleActivator {
         defaultAggregationProcessors.add(ParallelGroupedDataCountDistinctAggregationProcessor.getDefinition());
         defaultAggregationProcessors.add(ParallelGroupedDataCountAggregationProcessor.getDefinition());
         defaultAggregationProcessors.add(ParallelGroupedNumberDataAverageAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberPairDataAverageAggregationProcessor.getDefinition());
         defaultAggregationProcessors.add(ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition());
         defaultAggregationProcessors.add(ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition());
         defaultAggregationProcessors.add(ParallelGroupedNumberDataMinAggregationProcessor.getDefinition());

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberPairDataAverageAggregationProcessor.java

@@ -0,0 +1,101 @@
+package com.sap.sse.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.omg.CORBA.DoubleHolder;
+
+import com.sap.sse.common.Util.Pair;
+import com.sap.sse.datamining.components.AggregationProcessorDefinition;
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
+import com.sap.sse.datamining.shared.GroupKey;
+import com.sap.sse.datamining.shared.data.AverageWithStats;
+import com.sap.sse.datamining.shared.data.AveragePairWithStats;
+import com.sap.sse.datamining.shared.impl.AveragePairWithStatsImpl;
+import com.sap.sse.datamining.shared.impl.AverageWithStatsImpl;
+
+public class ParallelGroupedNumberPairDataAverageAggregationProcessor
+            extends AbstractParallelGroupedDataStoringAggregationProcessor<Pair, AveragePairWithStats<Number>> {
+    private static final Class<?> _c = AveragePairWithStats.class;
+    @SuppressWarnings("unchecked")
+    private static final Class<AveragePairWithStats<Number>> _cc = (Class<AveragePairWithStats<Number>>) _c;
+    
+    private static final AggregationProcessorDefinition<Pair, AveragePairWithStats<Number>> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Pair.class,
+                    _cc, "PairAverage", ParallelGroupedNumberPairDataAverageAggregationProcessor.class);
+    
+    public static AggregationProcessorDefinition<Pair, AveragePairWithStats<Number>> getDefinition() {
+        return DEFINITION;
+    }
+
+    private final Map<GroupKey, Pair<Number, Number>> sumPerKey;
+    private final Map<GroupKey, Pair<Number, Number>> minPerKey;
+    private final Map<GroupKey, Pair<Number, Number>> maxPerKey;
+    private final Map<GroupKey, AtomicLong> elementAmountPerKey;
+
+    public ParallelGroupedNumberPairDataAverageAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, AveragePairWithStats<Number>>, ?>> resultReceivers) {
+        super(executor, resultReceivers, "PairAverage");
+        elementAmountPerKey = new HashMap<>();
+        sumPerKey = new HashMap<>();
+        minPerKey = new HashMap<>();
+        maxPerKey = new HashMap<>();
+    }
+
+    @SuppressWarnings({ "unchecked", "rawtypes" })
+    @Override
+    protected void storeElement(GroupedDataEntry<Pair> element) {
+        if (element.getDataEntry() != null) {
+            incrementElementAmount(element);
+            // concurrency is not an issue here; needsSynchronization() returns true
+            Pair<Number, Number> aggregate = sumPerKey.get(element.getKey());
+            final double doubleValueA = ((Number) element.getDataEntry().getA()).doubleValue();
+            final double doubleValueB = ((Number) element.getDataEntry().getA()).doubleValue();
+            if (aggregate == null) {
+                aggregate = new Pair<>(doubleValueA, doubleValueB);
+                sumPerKey.put(element.getKey(), aggregate);
+            } else {
+                sumPerKey.put(element.getKey(), new Pair<>(((Number) element.getDataEntry().getA()).doubleValue() + doubleValueA,((Number) element.getDataEntry().getB()).doubleValue() + doubleValueB));
+            }
+            if (!minPerKey.containsKey(element.getKey()) || doubleValueA < minPerKey.get(element.getKey()).getA().doubleValue() && doubleValueB < minPerKey.get(element.getKey()).getB().doubleValue()) {
+                minPerKey.put(element.getKey(), new Pair<>(doubleValueA, doubleValueB));
+            }
+            if (!maxPerKey.containsKey(element.getKey()) || doubleValueA > maxPerKey.get(element.getKey()).getA().doubleValue() && doubleValueB > maxPerKey.get(element.getKey()).getB().doubleValue()) {
+                maxPerKey.put(element.getKey(), new Pair<>(doubleValueA, doubleValueB));
+            }
+        }
+    }
+
+    private void incrementElementAmount(GroupedDataEntry<Pair> element) {
+        GroupKey key = element.getKey();
+        // concurrency is not an issue here; needsSynchronization() returns true
+        AtomicLong currentAmount = elementAmountPerKey.get(key);
+        if (currentAmount == null) {
+            elementAmountPerKey.put(key, new AtomicLong(1));
+        } else {
+            currentAmount.incrementAndGet();
+        }
+    }
+
+    @Override
+    protected Map<GroupKey, AveragePairWithStats<Number>> aggregateResult() {
+        Map<GroupKey, AveragePairWithStats<Number>> result = new HashMap<>();
+        for (Entry<GroupKey, Pair<Number, Number>> sumAggregationEntry : sumPerKey.entrySet()) {
+            GroupKey key = sumAggregationEntry.getKey();
+            result.put(key, new AveragePairWithStatsImpl<Number>(new Pair<>(sumAggregationEntry.getValue().getA().doubleValue() / elementAmountPerKey.get(key).get(),sumAggregationEntry.getValue().getB().doubleValue() / elementAmountPerKey.get(key).get()),
+                    minPerKey.get(key), maxPerKey.get(key),
+                    /* median */ null,
+                    /* standardDeviation */ null,
+                    /* count */ elementAmountPerKey.get(key).get(),
+                    Number.class.getName()));
+        }
+        return result;
+    }
+
+}