Improved the aggregators

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/Activator.java

@@ -13,7 +13,7 @@ import com.sap.sailing.datamining.data.HasRaceResultOfCompetitorContext;
 import com.sap.sailing.datamining.data.HasTrackedLegContext;
 import com.sap.sailing.datamining.data.HasTrackedLegOfCompetitorContext;
 import com.sap.sailing.datamining.data.HasTrackedRaceContext;
-import com.sap.sailing.datamining.impl.components.aggregators.ParallelTrueSumAggregationProcessor;
+import com.sap.sailing.datamining.impl.components.aggregators.ParallelBoolSumAggregationProcessor;
 import com.sap.sailing.domain.common.Speed;
 import com.sap.sse.datamining.DataMiningBundleService;
 import com.sap.sse.datamining.DataSourceProvider;
@@ -98,7 +98,7 @@ public class Activator extends AbstractDataMiningActivator implements DataMining
     @Override
     public Iterable<AggregationProcessorDefinition<?, ?>> getAggregationProcessorDefinitions() {
         HashSet<AggregationProcessorDefinition<?, ?>> aggregators = new HashSet<>();
-        aggregators.add(ParallelTrueSumAggregationProcessor.getDefinition());
+        aggregators.add(ParallelBoolSumAggregationProcessor.getDefinition());
         return aggregators;
     }
     

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelBoolSumAggregationProcessor.java

@@ -9,14 +9,14 @@ 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.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public class ParallelTrueSumAggregationProcessor
-             extends AbstractParallelGroupedDataStoringAggregationProcessor<Boolean, Double> {
+public class ParallelBoolSumAggregationProcessor
+             extends AbstractParallelGroupedDataAggregationProcessor<Boolean, Double> {
     
     private static final AggregationProcessorDefinition<Boolean, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Boolean.class, Double.class, "Sum", ParallelTrueSumAggregationProcessor.class);
+            new SimpleAggregationProcessorDefinition<>(Boolean.class, Double.class, "Sum", ParallelBoolSumAggregationProcessor.class);
     
     public static AggregationProcessorDefinition<Boolean, Double> getDefinition() {
         return DEFINITION;
@@ -24,14 +24,14 @@ public class ParallelTrueSumAggregationProcessor
 
     private Map<GroupKey, Double> result;
 
-    public ParallelTrueSumAggregationProcessor(ExecutorService executor,
+    public ParallelBoolSumAggregationProcessor(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Sum");
         result = new HashMap<>();
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Boolean> element) {
+    protected void handleElement(GroupedDataEntry<Boolean> element) {
         GroupKey key = element.getKey();
         if (!result.containsKey(key)) {
             result.put(key, 0.0);
@@ -43,7 +43,7 @@ public class ParallelTrueSumAggregationProcessor
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Double> getResult() {
         return result;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestStatisticQuery.java

@@ -25,7 +25,7 @@ import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.data.QueryResult;
 import com.sap.sse.datamining.factories.FunctionFactory;
 import com.sap.sse.datamining.impl.components.AbstractRetrievalProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataSumAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataSumAggregationProcessor;
 import com.sap.sse.datamining.impl.data.QueryResultImpl;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.shared.data.QueryResultState;
@@ -56,9 +56,9 @@ public class TestStatisticQuery {
             }
         });
 
-        Query<Double> query = server.createQuery(createQueryDefinition());
+        Query<java.lang.Number> query = server.createQuery(createQueryDefinition());
         assertThat(query.getState(), is(QueryState.NOT_STARTED));
-        QueryResult<Double> expectedResult = buildExpectedResult(dataSource);
+        QueryResult<java.lang.Number> expectedResult = buildExpectedResult(dataSource);
         verifyResult(query.run(), expectedResult);
     }
 
@@ -66,7 +66,7 @@ public class TestStatisticQuery {
      * Creates a query definition, that filters all numbers < 10, groups them by
      * their length, extracts the cross sum and aggregates their sum.
      */
-    private StatisticQueryDefinition<Collection<Number>, Number, Double, Double> createQueryDefinition() {
+    private StatisticQueryDefinition<Collection<Number>, Number, java.lang.Number, java.lang.Number> createQueryDefinition() {
         FunctionFactory functionFactory = FunctionTestsUtil.getFunctionFactory();
         
         @SuppressWarnings("unchecked")
@@ -74,8 +74,8 @@ public class TestStatisticQuery {
         retrieverChain.startWith(NumberRetrievalProcessor.class, Number.class, "Number");
         
         Method getCrossSumMethod = FunctionTestsUtil.getMethodFromClass(Number.class, "getCrossSum");
-        AggregationProcessorDefinition<Double, Double> sumAggregatorDefinition = ParallelGroupedDoubleDataSumAggregationProcessor.getDefinition();
-        ModifiableStatisticQueryDefinition<Collection<Number>, Number, Double, Double> definition =
+        AggregationProcessorDefinition<java.lang.Number, java.lang.Number> sumAggregatorDefinition = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition();
+        ModifiableStatisticQueryDefinition<Collection<Number>, Number, java.lang.Number, java.lang.Number> definition =
                 new ModifiableStatisticQueryDefinition<>(Locale.ENGLISH, retrieverChain, functionFactory.createMethodWrappingFunction(getCrossSumMethod), sumAggregatorDefinition);
         
         Method getLengthMethod = FunctionTestsUtil.getMethodFromClass(Number.class, "getLength");
@@ -135,17 +135,17 @@ public class TestStatisticQuery {
         return dataSource;
     }
 
-    private QueryResult<Double> buildExpectedResult(Collection<Number> dataSource) {
-        Map<GroupKey, Double> results = new HashMap<>();
+    private QueryResult<java.lang.Number> buildExpectedResult(Collection<Number> dataSource) {
+        Map<GroupKey, java.lang.Number> results = new HashMap<>();
         results.put(new GenericGroupKey<Integer>(2), 5.0);
         results.put(new GenericGroupKey<Integer>(3), 3.0);
         results.put(new GenericGroupKey<Integer>(4), 10.0);
         
-        QueryResultImpl<Double> result = new QueryResultImpl<>(QueryResultState.NORMAL, Double.class, results, new AdditionalResultDataImpl(dataSource.size() - 2, "Cross Sum (Sum)", Unit.None, "", 0, 0));
+        QueryResultImpl<java.lang.Number> result = new QueryResultImpl<>(QueryResultState.NORMAL, java.lang.Number.class, results, new AdditionalResultDataImpl(dataSource.size() - 2, "Cross Sum (Sum)", Unit.None, "", 0, 0));
         return result;
     }
 
-    private void verifyResult(QueryResult<Double> result, QueryResult<Double> expectedResult) {
+    private void verifyResult(QueryResult<java.lang.Number> result, QueryResult<java.lang.Number> expectedResult) {
         assertThat("The result State isn't correct.", result.getState(), is(expectedResult.getState()));
         assertThat("Result values aren't correct.", result.getResults(), is(expectedResult.getResults()));
         assertThat("Retrieved data amount isn't correct.", result.getRetrievedDataAmount(), is(expectedResult.getRetrievedDataAmount()));

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java

@@ -5,29 +5,36 @@ import static org.junit.Assert.assertThat;
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 
 import org.junit.Before;
 import org.junit.Test;
 
 import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.shared.GroupKey;
+import com.sap.sse.datamining.shared.impl.GenericGroupKey;
 import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
 import com.sap.sse.datamining.test.util.components.NullProcessor;
 
 public class TestAbstractStoringParallelAggregationProcessor {
 
-    private Collection<Processor<Integer, ?>> receivers;
+    private Collection<Processor<Map<GroupKey, Integer>, ?>> receivers;
     private boolean receiverWasToldToFinish = false;
-    private Integer receivedElement = null;
+    private GroupKey groupKey = new GenericGroupKey<>("Key");
+    private Map<GroupKey, Integer> result = null;
     
-    private Collection<Integer> elementStore = new ArrayList<>();
+    private Collection<GroupedDataEntry<Integer>> elementStore = new ArrayList<>();
     
     @Before
     public void initializeReceivers() {
-        Processor<Integer, Void> receiver = new NullProcessor<Integer, Void>(Integer.class, Void.class) {
+        @SuppressWarnings("unchecked")
+        Processor<Map<GroupKey, Integer>, Void> receiver = new NullProcessor<Map<GroupKey,Integer>, Void>((Class<Map<GroupKey, Integer>>)(Class<?>) Map.class, Void.class) {
             @Override
-            public void processElement(Integer element) {
-                receivedElement = element;
+            public void processElement(Map<GroupKey, Integer> element) {
+                result = element;
             }
             @Override
             public void finish() throws InterruptedException {
@@ -41,43 +48,48 @@ public class TestAbstractStoringParallelAggregationProcessor {
 
     @Test
     public void testAbstractAggregationHandling() throws InterruptedException {
-        Processor<Integer, Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
+        Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
             @Override
-            protected void storeElement(Integer element) {
+            protected void storeElement(GroupedDataEntry<Integer> element) {
                 elementStore.add(element);
             }
             @Override
-            protected Integer aggregateResult() {
-                Integer sum = 0;
-                for (Integer element : elementStore) {
-                    sum += element;
+            protected Map<GroupKey, Integer> aggregateResult() {
+                Map<GroupKey, Integer> result = new HashMap<>();
+                for (GroupedDataEntry<Integer> element : elementStore) {
+                    GroupKey key = element.getKey();
+                    if (!result.containsKey(key)) {
+                        result.put(key, 0);
+                    }
+                    result.put(key, result.get(key) + element.getDataEntry());
                 }
-                return sum;
+                return result;
             }
         };
 
-        processElementAndVerifyThatItWasStored(processor, 42);
-        processElementAndVerifyThatItWasStored(processor, 7);
+        processElementAndVerifyThatItWasStored(processor, new GroupedDataEntry<>(groupKey, 42));
+        processElementAndVerifyThatItWasStored(processor, new GroupedDataEntry<>(groupKey, 7));
         
         processor.finish();
         assertThat("The receiver wasn't told to finish", receiverWasToldToFinish, is(true));
-        Integer expectedReceivedElement = 42 + 7;
-        assertThat(receivedElement, is(expectedReceivedElement));
+        int expectedData = 42 + 7;
+        assertThat(result.get(groupKey), is(expectedData));
     }
 
-    private void processElementAndVerifyThatItWasStored(Processor<Integer, Integer> processor, int element) {
+    private void processElementAndVerifyThatItWasStored(Processor<GroupedDataEntry<Integer>, ?> processor, GroupedDataEntry<Integer> element) {
         processor.processElement(element);
         ConcurrencyTestsUtil.sleepFor(100); //Giving the processor time to process the instructions
         assertThat("The element store doesn't contain the previously processed element '" + element + "'", elementStore.contains(element), is(true));
     }
     
+    @SuppressWarnings("unchecked")
     @Test(timeout=5000)
     public void testThatTheLockIsReleasedAfterStoringFailed() throws InterruptedException {
-        Collection<Processor<Integer, ?>> receivers = new HashSet<>();
-        receivers.add(new NullProcessor<Integer, Void>(Integer.class, Void.class) {
+        Collection<Processor<Map<GroupKey, Integer>, ?>> receivers = new HashSet<>();
+        receivers.add(new NullProcessor<Map<GroupKey, Integer>, Void>((Class<Map<GroupKey, Integer>>)(Class<?>) Map.class, Void.class) {
             @Override
-            public void processElement(Integer element) {
-                receivedElement = element;
+            public void processElement(Map<GroupKey, Integer> element) {
+                result = element;
             }
             @Override
             public void finish() throws InterruptedException {
@@ -92,32 +104,36 @@ public class TestAbstractStoringParallelAggregationProcessor {
                 }
             }
         });
-        Processor<Integer, Integer> processor = new AbstractParallelStoringAggregationProcessor<Integer, Integer>(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
+        Processor<GroupedDataEntry<Integer>, Map<GroupKey, Integer>> processor = new AbstractParallelGroupedDataStoringAggregationProcessor<Integer, Integer>(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") {
             @Override
-            protected void storeElement(Integer element) {
-                if (element < 0) {
+            protected void storeElement(GroupedDataEntry<Integer> element) {
+                if (element.getDataEntry() < 0) {
                     throw new IllegalArgumentException("The element mustn't be negative");
                 }
                 elementStore.add(element);
             }
             @Override
-            protected Integer aggregateResult() {
-                Integer sum = 0;
-                for (Integer element : elementStore) {
-                    sum += element;
+            protected Map<GroupKey, Integer> aggregateResult() {
+                Map<GroupKey, Integer> result = new HashMap<>();
+                for (GroupedDataEntry<Integer> element : elementStore) {
+                    GroupKey key = element.getKey();
+                    if (!result.containsKey(key)) {
+                        result.put(key, 0);
+                    }
+                    result.put(key, result.get(key) + element.getDataEntry());
                 }
-                return sum;
+                return result;
             }
         };
 
-        processor.processElement(-1);
-        processor.processElement(42);
-        processor.processElement(7);
+        processor.processElement(new GroupedDataEntry<>(groupKey, -1));
+        processor.processElement(new GroupedDataEntry<>(groupKey, 42));
+        processor.processElement(new GroupedDataEntry<>(groupKey, 7));
         
         processor.finish();
         assertThat("The receiver wasn't told to finish", receiverWasToldToFinish, is(true));
-        Integer expectedReceivedElement = 42 + 7;
-        assertThat(receivedElement, is(expectedReceivedElement));
+        int expectedData = 42 + 7;
+        assertThat(result.get(groupKey), is(expectedData));
     }
 
 }

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelAggregationProcessors.java

@@ -16,61 +16,61 @@ import com.sap.sse.datamining.shared.impl.GenericGroupKey;
 import com.sap.sse.datamining.test.util.ConcurrencyTestsUtil;
 import com.sap.sse.datamining.test.util.components.NullProcessor;
 
-public class TestParallelDoubleAggregationProcessors {
+public class TestParallelAggregationProcessors {
     
-    private Collection<Processor<Map<GroupKey, Double>, ?>> receivers;
-    private Map<GroupKey, Double> receivedAggregations;
+    private Collection<Processor<Map<GroupKey, Number>, ?>> receivers;
+    private Map<GroupKey, Number> receivedAggregations;
 
     @Test
     public void testSumAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> sumAggregationProcessor = ParallelGroupedDoubleDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
-        Collection<GroupedDataEntry<Double>> elements = createElements();
+        Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> sumAggregationProcessor = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Collection<GroupedDataEntry<Number>> elements = createElements();
         ConcurrencyTestsUtil.processElements(sumAggregationProcessor, elements);
         
         sumAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedSumAggregations(elements);
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedSumAggregations(elements);
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedSumAggregations(Collection<GroupedDataEntry<Double>> elements) {
-        Map<GroupKey, Double> expectedSumAggregations = new HashMap<>();
-        for (GroupedDataEntry<Double> element : elements) {
+    private Map<GroupKey, Number> computeExpectedSumAggregations(Collection<GroupedDataEntry<Number>> elements) {
+        Map<GroupKey, Number> expectedSumAggregations = new HashMap<>();
+        for (GroupedDataEntry<Number> element : elements) {
             GroupKey key = element.getKey();
             if (!expectedSumAggregations.containsKey(key)) {
                 expectedSumAggregations.put(key, 0.0);
             }
-            Double currentValue = expectedSumAggregations.get(key);
-            expectedSumAggregations.put(key, currentValue + element.getDataEntry());
+            Number currentValue = expectedSumAggregations.get(key);
+            expectedSumAggregations.put(key, currentValue.doubleValue() + element.getDataEntry().doubleValue());
         }
         return expectedSumAggregations;
     }
 
     @Test
     public void testAverageAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> averageAggregationProcessor = ParallelGroupedDoubleDataAverageAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
-        Collection<GroupedDataEntry<Double>> elements = createElements();
+        Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> averageAggregationProcessor = ParallelGroupedNumberDataAverageAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Collection<GroupedDataEntry<Number>> elements = createElements();
         ConcurrencyTestsUtil.processElements(averageAggregationProcessor, elements);
         
         averageAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedAverageAggregations(elements);
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedAverageAggregations(elements);
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedAverageAggregations(Collection<GroupedDataEntry<Double>> elements) {
-        Map<GroupKey, Double> result = new HashMap<>();
+    private Map<GroupKey, Number> computeExpectedAverageAggregations(Collection<GroupedDataEntry<Number>> elements) {
+        Map<GroupKey, Number> result = new HashMap<>();
         
-        Map<GroupKey, Double> sumAggregations = computeExpectedSumAggregations(elements);
+        Map<GroupKey, Number> sumAggregations = computeExpectedSumAggregations(elements);
         Map<GroupKey, Double> elementAmountPerKey = countElementAmountPerKey(elements);
-        for (Entry<GroupKey, Double> sumAggregationEntry : sumAggregations.entrySet()) {
+        for (Entry<GroupKey, Number> sumAggregationEntry : sumAggregations.entrySet()) {
             GroupKey key = sumAggregationEntry.getKey();
-            result.put(key, sumAggregationEntry.getValue() / elementAmountPerKey.get(key));
+            result.put(key, sumAggregationEntry.getValue().doubleValue() / elementAmountPerKey.get(key));
         }
         return result;
     }
 
-    private Map<GroupKey, Double> countElementAmountPerKey(Collection<GroupedDataEntry<Double>> elements) {
+    private Map<GroupKey, Double> countElementAmountPerKey(Collection<GroupedDataEntry<Number>> elements) {
         Map<GroupKey, Double> elementAmountPerKey = new HashMap<>();
-        for (GroupedDataEntry<Double> element : elements) {
+        for (GroupedDataEntry<?> element : elements) {
             GroupKey key = element.getKey();
             if (!elementAmountPerKey.containsKey(key)) {
                 elementAmountPerKey.put(key, 0.0);
@@ -83,17 +83,17 @@ public class TestParallelDoubleAggregationProcessors {
 
     @Test
     public void testMedianAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> medianAggregationProcessor = ParallelGroupedDoubleDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
-        Collection<GroupedDataEntry<Double>> elements = createElements();
+        Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> medianAggregationProcessor = ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Collection<GroupedDataEntry<Number>> elements = createElements();
         ConcurrencyTestsUtil.processElements(medianAggregationProcessor, elements);
         
         medianAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedMedianAggregations();
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedMedianAggregations();
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedMedianAggregations() {
-        Map<GroupKey, Double> result = new HashMap<>();
+    private Map<GroupKey, Number> computeExpectedMedianAggregations() {
+        Map<GroupKey, Number> result = new HashMap<>();
         result.put(new GenericGroupKey<Integer>(1), 7.0);
         result.put(new GenericGroupKey<Integer>(2), 6.0);
         result.put(new GenericGroupKey<Integer>(3), 5.0);
@@ -102,17 +102,17 @@ public class TestParallelDoubleAggregationProcessors {
 
     @Test
     public void testMaxAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> maxAggregationProcessor = ParallelGroupedDoubleDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
-        Collection<GroupedDataEntry<Double>> elements = createElements();
+        Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> maxAggregationProcessor = ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Collection<GroupedDataEntry<Number>> elements = createElements();
         ConcurrencyTestsUtil.processElements(maxAggregationProcessor, elements);
         
         maxAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedMaxAggregations();
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedMaxAggregations();
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedMaxAggregations() {
-        Map<GroupKey, Double> result = new HashMap<>();
+    private Map<GroupKey, Number> computeExpectedMaxAggregations() {
+        Map<GroupKey, Number> result = new HashMap<>();
         result.put(new GenericGroupKey<Integer>(1), 10.0);
         result.put(new GenericGroupKey<Integer>(2), 7.0);
         result.put(new GenericGroupKey<Integer>(3), 5.0);
@@ -121,17 +121,17 @@ public class TestParallelDoubleAggregationProcessors {
 
     @Test
     public void testMinAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Double>, Map<GroupKey, Double>> minAggregationProcessor = ParallelGroupedDoubleDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
-        Collection<GroupedDataEntry<Double>> elements = createElements();
+        Processor<GroupedDataEntry<Number>, Map<GroupKey, Number>> minAggregationProcessor = ParallelGroupedNumberDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Collection<GroupedDataEntry<Number>> elements = createElements();
         ConcurrencyTestsUtil.processElements(minAggregationProcessor, elements);
         
         minAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedMinAggregations();
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedMinAggregations();
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedMinAggregations() {
-        Map<GroupKey, Double> result = new HashMap<>();
+    private Map<GroupKey, Number> computeExpectedMinAggregations() {
+        Map<GroupKey, Number> result = new HashMap<>();
         result.put(new GenericGroupKey<Integer>(1), 5.0);
         result.put(new GenericGroupKey<Integer>(2), 3.0);
         result.put(new GenericGroupKey<Integer>(3), 5.0);
@@ -140,42 +140,42 @@ public class TestParallelDoubleAggregationProcessors {
 
     @Test
     public void testCountAggregationProcessor() throws InterruptedException {
-        Processor<GroupedDataEntry<Object>, Map<GroupKey, Double>> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
+        Processor<GroupedDataEntry<Object>, Map<GroupKey, Number>> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers);
         @SuppressWarnings("unchecked")
         Collection<GroupedDataEntry<Object>> elements = (Collection<GroupedDataEntry<Object>>)(Collection<?>) createElements();
         ConcurrencyTestsUtil.processElements(countAggregationProcessor, elements);
         
         countAggregationProcessor.finish();
-        Map<GroupKey, Double> expectedReceivedAggregations = computeExpectedCountAggregations();
+        Map<GroupKey, Number> expectedReceivedAggregations = computeExpectedCountAggregations();
         ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations);
     }
 
-    private Map<GroupKey, Double> computeExpectedCountAggregations() {
-        Map<GroupKey, Double> result = new HashMap<>();
+    private Map<GroupKey, Number> computeExpectedCountAggregations() {
+        Map<GroupKey, Number> result = new HashMap<>();
         result.put(new GenericGroupKey<Integer>(1), 3.0);
         result.put(new GenericGroupKey<Integer>(2), 4.0);
         result.put(new GenericGroupKey<Integer>(3), 3.0);
         return result;
     }
 
-    private Collection<GroupedDataEntry<Double>> createElements() {
-        Collection<GroupedDataEntry<Double>> elements = new ArrayList<>();
+    private Collection<GroupedDataEntry<Number>> createElements() {
+        Collection<GroupedDataEntry<Number>> elements = new ArrayList<>();
         
         GroupKey firstGroupKey = new GenericGroupKey<Integer>(1);
-        elements.add(new GroupedDataEntry<Double>(firstGroupKey, 5.0));
-        elements.add(new GroupedDataEntry<Double>(firstGroupKey, 10.0));
-        elements.add(new GroupedDataEntry<Double>(firstGroupKey, 7.0));
+        elements.add(new GroupedDataEntry<Number>(firstGroupKey, 5.0));
+        elements.add(new GroupedDataEntry<Number>(firstGroupKey, 10.0));
+        elements.add(new GroupedDataEntry<Number>(firstGroupKey, 7.0));
         
         GroupKey secondGroupKey = new GenericGroupKey<Integer>(2);
-        elements.add(new GroupedDataEntry<Double>(secondGroupKey, 5.0));
-        elements.add(new GroupedDataEntry<Double>(secondGroupKey, 3.0));
-        elements.add(new GroupedDataEntry<Double>(secondGroupKey, 7.0));
-        elements.add(new GroupedDataEntry<Double>(secondGroupKey, 7.0));
+        elements.add(new GroupedDataEntry<Number>(secondGroupKey, 5.0));
+        elements.add(new GroupedDataEntry<Number>(secondGroupKey, 3.0));
+        elements.add(new GroupedDataEntry<Number>(secondGroupKey, 7.0));
+        elements.add(new GroupedDataEntry<Number>(secondGroupKey, 7.0));
         
         GroupKey thirdGroupKey = new GenericGroupKey<Integer>(3);
-        elements.add(new GroupedDataEntry<Double>(thirdGroupKey, 5.0));
-        elements.add(new GroupedDataEntry<Double>(thirdGroupKey, 5.0));
-        elements.add(new GroupedDataEntry<Double>(thirdGroupKey, 5.0));
+        elements.add(new GroupedDataEntry<Number>(thirdGroupKey, 5.0));
+        elements.add(new GroupedDataEntry<Number>(thirdGroupKey, 5.0));
+        elements.add(new GroupedDataEntry<Number>(thirdGroupKey, 5.0));
         
         return elements;
     }
@@ -183,9 +183,9 @@ public class TestParallelDoubleAggregationProcessors {
     @Before
     public void initializeResultReceivers() {
         @SuppressWarnings("unchecked")
-        Processor<Map<GroupKey, Double>, Void> receiver = new NullProcessor<Map<GroupKey, Double>, Void>((Class<Map<GroupKey, Double>>)(Class<?>) Map.class, Void.class) {
+        Processor<Map<GroupKey, Number>, Void> receiver = new NullProcessor<Map<GroupKey, Number>, Void>((Class<Map<GroupKey, Number>>)(Class<?>) Map.class, Void.class) {
             @Override
-            public void processElement(Map<GroupKey, Double> element) {
+            public void processElement(Map<GroupKey, Number> element) {
                 receivedAggregations = element;
             }
         };

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAggregationProcessorDummyWithInvalidConstructor.java

@@ -6,11 +6,11 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public class TestAggregationProcessorDummyWithInvalidConstructor extends
-        AbstractParallelGroupedDataStoringAggregationProcessor<Object, Double> {
+        AbstractParallelGroupedDataAggregationProcessor<Object, Double> {
 
     public TestAggregationProcessorDummyWithInvalidConstructor(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers, Object constructorInvalidator) {
@@ -18,11 +18,11 @@ public class TestAggregationProcessorDummyWithInvalidConstructor extends
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Object> element) {
+    protected void handleElement(GroupedDataEntry<Object> element) {
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Double> getResult() {
         return null;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAnyDataTypeAggregationProcessorDummy.java

@@ -6,11 +6,11 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public class TestAnyDataTypeAggregationProcessorDummy extends
-        AbstractParallelGroupedDataStoringAggregationProcessor<Object, Double> {
+        AbstractParallelGroupedDataAggregationProcessor<Object, Double> {
 
     public TestAnyDataTypeAggregationProcessorDummy(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
@@ -18,11 +18,11 @@ public class TestAnyDataTypeAggregationProcessorDummy extends
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Object> element) {
+    protected void handleElement(GroupedDataEntry<Object> element) {
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Double> getResult() {
         return null;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestCompetitorAggregationProcessorDummy.java

@@ -6,12 +6,12 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.test.domain.Test_Competitor;
 
 public class TestCompetitorAggregationProcessorDummy extends
-        AbstractParallelGroupedDataStoringAggregationProcessor<Test_Competitor, Test_Competitor> {
+        AbstractParallelGroupedDataAggregationProcessor<Test_Competitor, Test_Competitor> {
 
     public TestCompetitorAggregationProcessorDummy(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Test_Competitor>, ?>> resultReceivers) {
@@ -19,11 +19,11 @@ public class TestCompetitorAggregationProcessorDummy extends
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Test_Competitor> element) {
+    protected void handleElement(GroupedDataEntry<Test_Competitor> element) {
     }
 
     @Override
-    protected Map<GroupKey, Test_Competitor> aggregateResult() {
+    protected Map<GroupKey, Test_Competitor> getResult() {
         return null;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestNamedAggregationProcessorDummy.java

@@ -6,12 +6,12 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.test.domain.Test_Named;
 
 public class TestNamedAggregationProcessorDummy extends
-        AbstractParallelGroupedDataStoringAggregationProcessor<Test_Named, Test_Named> {
+        AbstractParallelGroupedDataAggregationProcessor<Test_Named, Test_Named> {
 
     public TestNamedAggregationProcessorDummy(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Test_Named>, ?>> resultReceivers) {
@@ -19,11 +19,11 @@ public class TestNamedAggregationProcessorDummy extends
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Test_Named> element) {
+    protected void handleElement(GroupedDataEntry<Test_Named> element) {
     }
 
     @Override
-    protected Map<GroupKey, Test_Named> aggregateResult() {
+    protected Map<GroupKey, Test_Named> getResult() {
         return null;
     }
 

java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestRaceAggregationProcessorDummy.java

@@ -6,12 +6,12 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.test.domain.Test_Race;
 
 public class TestRaceAggregationProcessorDummy extends
-        AbstractParallelGroupedDataStoringAggregationProcessor<Test_Race, Test_Race> {
+        AbstractParallelGroupedDataAggregationProcessor<Test_Race, Test_Race> {
 
     public TestRaceAggregationProcessorDummy(ExecutorService executor,
             Collection<Processor<Map<GroupKey, Test_Race>, ?>> resultReceivers) {
@@ -19,11 +19,11 @@ public class TestRaceAggregationProcessorDummy extends
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Test_Race> element) {
+    protected void handleElement(GroupedDataEntry<Test_Race> element) {
     }
 
     @Override
-    protected Map<GroupKey, Test_Race> aggregateResult() {
+    protected Map<GroupKey, Test_Race> getResult() {
         return null;
     }
 

java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/AggregationProcessorDefinition.java

@@ -5,7 +5,7 @@ import java.util.Map;
 import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public interface AggregationProcessorDefinition<ExtractedType, AggregatedType> {
@@ -15,7 +15,7 @@ public interface AggregationProcessorDefinition<ExtractedType, AggregatedType> {
     
     public String getAggregationNameMessageKey();
     
-    public Class<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> getAggregationProcessor();
+    public Class<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> getAggregationProcessor();
     public Processor<GroupedDataEntry<ExtractedType>, Map<GroupKey, AggregatedType>> construct(ExecutorService executor, Collection<Processor<Map<GroupKey, AggregatedType>, ?>> resultReceivers);
     
 }

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

@@ -21,11 +21,11 @@ import com.sap.sse.datamining.components.DataRetrieverChainDefinition;
 import com.sap.sse.datamining.components.management.AggregationProcessorDefinitionRegistry;
 import com.sap.sse.datamining.components.management.DataRetrieverChainDefinitionRegistry;
 import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDataCountAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataAverageAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataMaxAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataMedianAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataMinAggregationProcessor;
-import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedDoubleDataSumAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataAverageAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.ParallelGroupedNumberDataMaxAggregationProcessor;
+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.management.AggregationProcessorDefinitionManager;
 import com.sap.sse.datamining.impl.components.management.DataRetrieverChainDefinitionManager;
 import com.sap.sse.datamining.impl.functions.FunctionManager;
@@ -64,11 +64,11 @@ public class DataMiningFrameworkActivator implements BundleActivator {
     private Iterable<AggregationProcessorDefinition<?, ?>> getDefaultAggregationProcessors() {
         Collection<AggregationProcessorDefinition<?, ?>> defaultAggregationProcessors = new HashSet<>();
         defaultAggregationProcessors.add(ParallelGroupedDataCountAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedDoubleDataAverageAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedDoubleDataMaxAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedDoubleDataMedianAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedDoubleDataMinAggregationProcessor.getDefinition());
-        defaultAggregationProcessors.add(ParallelGroupedDoubleDataSumAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberDataAverageAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberDataMinAggregationProcessor.getDefinition());
+        defaultAggregationProcessors.add(ParallelGroupedNumberDataSumAggregationProcessor.getDefinition());
         return defaultAggregationProcessors;
     }
 

java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/SimpleAggregationProcessorDefinition.java

@@ -9,7 +9,7 @@ import java.util.concurrent.ExecutorService;
 
 import com.sap.sse.datamining.components.AggregationProcessorDefinition;
 import com.sap.sse.datamining.components.Processor;
-import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataStoringAggregationProcessor;
+import com.sap.sse.datamining.impl.components.aggregators.AbstractParallelGroupedDataAggregationProcessor;
 import com.sap.sse.datamining.shared.GroupKey;
 
 public class SimpleAggregationProcessorDefinition<ExtractedType, AggregatedType> implements
@@ -18,14 +18,14 @@ public class SimpleAggregationProcessorDefinition<ExtractedType, AggregatedType>
     private final Class<ExtractedType> extractedType;
     private final Class<AggregatedType> aggregatedType;
     private final String aggregationNameMessageKey;
-    private final Class<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor;
-    private final Constructor<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> constructor;
+    private final Class<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor;
+    private final Constructor<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> constructor;
     
     public SimpleAggregationProcessorDefinition(
             Class<ExtractedType> extractedType,
             Class<AggregatedType> aggregatedType,
             String aggregationNameMessageKey,
-            Class<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor) {
+            Class<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor) {
         this.extractedType = extractedType;
         this.aggregatedType = aggregatedType;
         this.aggregationNameMessageKey = aggregationNameMessageKey;
@@ -33,9 +33,9 @@ public class SimpleAggregationProcessorDefinition<ExtractedType, AggregatedType>
         constructor = ensureValidConstructor(aggregationProcessor);
     }
 
-    private Constructor<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> ensureValidConstructor(Class<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor) {
+    private Constructor<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> ensureValidConstructor(Class<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> aggregationProcessor) {
         try {
-            Constructor<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> constructor = aggregationProcessor.getConstructor(ExecutorService.class, Collection.class);
+            Constructor<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> constructor = aggregationProcessor.getConstructor(ExecutorService.class, Collection.class);
             if (Modifier.isPublic(constructor.getModifiers())) {
                 // Preventing IllegalAccessExceptions of public constructors due to weird package behaviour
                 constructor.setAccessible(true);
@@ -63,7 +63,7 @@ public class SimpleAggregationProcessorDefinition<ExtractedType, AggregatedType>
     }
 
     @Override
-    public Class<? extends AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>> getAggregationProcessor() {
+    public Class<? extends AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>> getAggregationProcessor() {
         return aggregationProcessor;
     }
     

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

@@ -11,20 +11,20 @@ import com.sap.sse.datamining.impl.components.AbstractParallelProcessor;
 import com.sap.sse.datamining.impl.components.AbstractProcessorInstruction;
 import com.sap.sse.datamining.impl.components.ProcessorInstructionPriority;
 
-public abstract class AbstractParallelStoringAggregationProcessor<InputType, AggregatedType> 
+public abstract class AbstractParallelAggregationProcessor<InputType, AggregatedType> 
                       extends AbstractParallelProcessor<InputType, AggregatedType> {
 
-    private final Lock storeLock;
     private final String aggregationNameMessageKey;
+    private final Lock writeLock;
 
-    public AbstractParallelStoringAggregationProcessor(Class<InputType> inputType,
-                                                       Class<AggregatedType> resultType,
-                                                       ExecutorService executor,
-                                                       Collection<Processor<AggregatedType, ?>> resultReceivers,
-                                                       String aggregationNameMessageKey) {
+    public AbstractParallelAggregationProcessor(Class<InputType> inputType,
+                                                Class<AggregatedType> resultType,
+                                                ExecutorService executor,
+                                                Collection<Processor<AggregatedType, ?>> resultReceivers,
+                                                String aggregationNameMessageKey) {
         super(inputType, resultType, executor, resultReceivers);
-        storeLock = new ReentrantLock();
         this.aggregationNameMessageKey = aggregationNameMessageKey;
+        writeLock = new ReentrantLock();
     }
 
     @Override
@@ -32,31 +32,32 @@ public abstract class AbstractParallelStoringAggregationProcessor<InputType, Agg
         return new AbstractProcessorInstruction<AggregatedType>(this, ProcessorInstructionPriority.Aggregation) {
             @Override
             public AggregatedType computeResult() {
-                storeLock.lock();
+                writeLock.lock();
                 try {
-                    storeElement(element);
+                    handleElement(element);
                 } finally {
-                    storeLock.unlock();
+                    writeLock.unlock();
                 }
-                return AbstractParallelStoringAggregationProcessor.super.createInvalidResult();
+                return AbstractParallelAggregationProcessor.super.createInvalidResult();
             }
         };
     }
 
+
     /**
-     * Method to store the element in the concrete store. This method is only called in a way, that is thread safe, so
-     * that multiple threads can't corrupt the store.
+     * Method to handle the element. This method is only called in a way, that is thread safe, so
+     * that multiple threads can't corrupt the data.
      */
-    protected abstract void storeElement(InputType element);
+    protected abstract void handleElement(InputType element);
     
     @Override
     public void finish() throws InterruptedException {
         super.sleepUntilAllInstructionsFinished();
-        super.forwardResultToReceivers(aggregateResult());
+        super.forwardResultToReceivers(getResult());
         super.tellResultReceiversToFinish();
     }
 
-    protected abstract AggregatedType aggregateResult();
+    protected abstract AggregatedType getResult();
     
     @Override
     protected void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder) {

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

@@ -0,0 +1,20 @@
+package com.sap.sse.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+import com.sap.sse.datamining.components.Processor;
+import com.sap.sse.datamining.impl.components.GroupedDataEntry;
+import com.sap.sse.datamining.shared.GroupKey;
+
+public abstract class AbstractParallelGroupedDataAggregationProcessor<ExtractedType, AggregatedType>
+                extends AbstractParallelAggregationProcessor<GroupedDataEntry<ExtractedType>, Map<GroupKey, AggregatedType>> {
+
+    @SuppressWarnings("unchecked")
+    public AbstractParallelGroupedDataAggregationProcessor(ExecutorService executor, Collection<Processor<Map<GroupKey, AggregatedType>, ?>> resultReceivers, String aggregationNameMessageKey) {
+        super((Class<GroupedDataEntry<ExtractedType>>)(Class<?>) GroupedDataEntry.class, (Class<Map<GroupKey, AggregatedType>>)(Class<?>) Map.class,
+                executor, resultReceivers, aggregationNameMessageKey);
+    }
+
+}

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

@@ -8,13 +8,31 @@ import com.sap.sse.datamining.components.Processor;
 import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public abstract class AbstractParallelGroupedDataStoringAggregationProcessor<ExtractedType, AggregatedType>
-                extends AbstractParallelStoringAggregationProcessor<GroupedDataEntry<ExtractedType>, Map<GroupKey, AggregatedType>> {
+public abstract class AbstractParallelGroupedDataStoringAggregationProcessor<InputType, AggregatedType> 
+                      extends AbstractParallelGroupedDataAggregationProcessor<InputType, AggregatedType> {
 
-    @SuppressWarnings("unchecked")
-    public AbstractParallelGroupedDataStoringAggregationProcessor(ExecutorService executor, Collection<Processor<Map<GroupKey, AggregatedType>, ?>> resultReceivers, String aggregationNameMessageKey) {
-        super((Class<GroupedDataEntry<ExtractedType>>)(Class<?>) GroupedDataEntry.class, (Class<Map<GroupKey, AggregatedType>>)(Class<?>) Map.class,
-                executor, resultReceivers, aggregationNameMessageKey);
+    public AbstractParallelGroupedDataStoringAggregationProcessor(ExecutorService executor,
+                                                                  Collection<Processor<Map<GroupKey, AggregatedType>, ?>> resultReceivers,
+                                                                  String aggregationNameMessageKey) {
+        super(executor, resultReceivers, aggregationNameMessageKey);
     }
 
+    @Override
+    protected void handleElement(GroupedDataEntry<InputType> element) {
+        storeElement(element);
+    }
+
+    /**
+     * Method to store the element in the concrete store. This method is only called in a way, that is thread safe, so
+     * that multiple threads can't corrupt the store.
+     */
+    protected abstract void storeElement(GroupedDataEntry<InputType> element);
+    
+    @Override
+    protected Map<GroupKey, AggregatedType> getResult() {
+        return aggregateResult();
+    }
+
+    protected abstract Map<GroupKey, AggregatedType> aggregateResult();
+
 }

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

@@ -14,35 +14,35 @@ import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.datamining.shared.data.Unit;
 
 public class ParallelGroupedDataCountAggregationProcessor
-                extends AbstractParallelGroupedDataStoringAggregationProcessor<Object, Double> {
+                extends AbstractParallelGroupedDataAggregationProcessor<Object, Number> {
     
-    private static final AggregationProcessorDefinition<Object, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Object.class, Double.class, "Count", ParallelGroupedDataCountAggregationProcessor.class);
+    private static final AggregationProcessorDefinition<Object, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Object.class, Number.class, "Count", ParallelGroupedDataCountAggregationProcessor.class);
     
-    public static AggregationProcessorDefinition<Object, Double> getDefinition() {
+    public static AggregationProcessorDefinition<Object, Number> getDefinition() {
         return DEFINITION;
     }
 
-    private Map<GroupKey, Double> countMap;
+    private Map<GroupKey, Number> countMap;
     
     public ParallelGroupedDataCountAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Count");
         countMap = new HashMap<>();
     }
-
+    
     @Override
-    protected void storeElement(GroupedDataEntry<Object> element) {
+    protected void handleElement(GroupedDataEntry<Object> element) {
         GroupKey key = element.getKey();
         if (!countMap.containsKey(key)) {
             countMap.put(key, 0.0);
         }
-        Double currentAmount = countMap.get(key);
+        Double currentAmount = countMap.get(key).doubleValue();
         countMap.put(key, currentAmount + 1.0);
     }
-
+    
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Number> getResult() {
         return countMap;
     }
     

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

@@ -1,72 +0,0 @@
-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 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;
-
-public class ParallelGroupedDoubleDataSumAggregationProcessor
-             extends AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> {
-    
-    private static final AggregationProcessorDefinition<Double, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Sum", ParallelGroupedDoubleDataSumAggregationProcessor.class);
-    
-    public static AggregationProcessorDefinition<Double, Double> getDefinition() {
-        return DEFINITION;
-    }
-
-    private Map<GroupedDataEntry<Double>, Integer> elementAmountMap;
-
-    public ParallelGroupedDoubleDataSumAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
-        super(executor, resultReceivers, "Sum");
-        elementAmountMap = new HashMap<>();
-    }
-
-    @Override
-    protected void storeElement(GroupedDataEntry<Double> element) {
-        if (!elementAmountMap.containsKey(element)) {
-            elementAmountMap.put(element, 0);
-        }
-        Integer currentAmount = elementAmountMap.get(element);
-        elementAmountMap.put(element, currentAmount + 1);
-    }
-
-    @Override
-    protected Map<GroupKey, Double> aggregateResult() {
-        Map<GroupKey, Double> result = new HashMap<>();
-        for (Entry<GroupedDataEntry<Double>, Integer> elementAmountEntry : elementAmountMap.entrySet()) {
-            Number element = elementAmountEntry.getKey().getDataEntry();
-            Integer times = elementAmountEntry.getValue();
-            Double multipliedElementValue = multiply(element.doubleValue(), times);
-            
-            GroupKey groupKey = elementAmountEntry.getKey().getKey();
-            Double groupResult = result.get(groupKey);
-            result.put(groupKey, addToGroupResult(groupResult, multipliedElementValue));
-        }
-        return result;
-    }
-
-    private Double multiply(Double element, Integer times) {
-        return element * times;
-    }
-
-    private Double add(Double firstSummand, Double secondSummand) {
-        return firstSummand + secondSummand;
-    }
-
-    private Double addToGroupResult(Double groupResult, Double multipliedElementValue) {
-        if (groupResult == null) {
-            return multipliedElementValue;
-        }
-        return add(groupResult, multipliedElementValue);
-    }
-
-}

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

@@ -12,33 +12,33 @@ import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public class ParallelGroupedDoubleDataAverageAggregationProcessor
-            extends AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> {
+public class ParallelGroupedNumberDataAverageAggregationProcessor
+            extends AbstractParallelGroupedDataStoringAggregationProcessor<Number, Number> {
     
-    private static final AggregationProcessorDefinition<Double, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Average", ParallelGroupedDoubleDataAverageAggregationProcessor.class);
+    private static final AggregationProcessorDefinition<Number, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Average", ParallelGroupedNumberDataAverageAggregationProcessor.class);
     
-    public static AggregationProcessorDefinition<Double, Double> getDefinition() {
+    public static AggregationProcessorDefinition<Number, Number> getDefinition() {
         return DEFINITION;
     }
 
-    private final AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> sumAggregationProcessor;
+    private final ParallelGroupedNumberDataSumAggregationProcessor sumAggregationProcessor;
     private final Map<GroupKey, Integer> elementAmountPerKey;
 
-    public ParallelGroupedDoubleDataAverageAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
+    public ParallelGroupedNumberDataAverageAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Average");
         elementAmountPerKey = new HashMap<>();
-        sumAggregationProcessor = new ParallelGroupedDoubleDataSumAggregationProcessor(executor, resultReceivers);
+        sumAggregationProcessor = new ParallelGroupedNumberDataSumAggregationProcessor(executor, resultReceivers);
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Double> element) {
+    protected void storeElement(GroupedDataEntry<Number> element) {
         incrementElementAmount(element);
-        sumAggregationProcessor.storeElement(element);
+        sumAggregationProcessor.handleElement(element);
     }
 
-    private void incrementElementAmount(GroupedDataEntry<Double> element) {
+    private void incrementElementAmount(GroupedDataEntry<Number> element) {
         GroupKey key = element.getKey();
         if (!elementAmountPerKey.containsKey(key)) {
             elementAmountPerKey.put(key, 0);
@@ -48,12 +48,12 @@ public class ParallelGroupedDoubleDataAverageAggregationProcessor
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
-        Map<GroupKey, Double> result = new HashMap<>();
-        Map<GroupKey, Double> sumAggregation = sumAggregationProcessor.aggregateResult();
-        for (Entry<GroupKey, Double> sumAggregationEntry : sumAggregation.entrySet()) {
+    protected Map<GroupKey, Number> aggregateResult() {
+        Map<GroupKey, Number> result = new HashMap<>();
+        Map<GroupKey, Number> sumAggregation = sumAggregationProcessor.getResult();
+        for (Entry<GroupKey, Number> sumAggregationEntry : sumAggregation.entrySet()) {
             GroupKey key = sumAggregationEntry.getKey();
-            result.put(key, sumAggregationEntry.getValue() / elementAmountPerKey.get(key));
+            result.put(key, sumAggregationEntry.getValue().doubleValue() / elementAmountPerKey.get(key).doubleValue());
         }
         return result;
     }

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

@@ -11,35 +11,35 @@ import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public class ParallelGroupedDoubleDataMaxAggregationProcessor
-                extends AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> {
+public class ParallelGroupedNumberDataMaxAggregationProcessor
+                extends AbstractParallelGroupedDataAggregationProcessor<Number, Number> {
     
-    private static final AggregationProcessorDefinition<Double, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Maximum", ParallelGroupedDoubleDataMaxAggregationProcessor.class);
+    private static final AggregationProcessorDefinition<Number, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Maximum", ParallelGroupedNumberDataMaxAggregationProcessor.class);
     
-    public static AggregationProcessorDefinition<Double, Double> getDefinition() {
+    public static AggregationProcessorDefinition<Number, Number> getDefinition() {
         return DEFINITION;
     }
 
-    private Map<GroupKey, Double> maxMap;
+    private Map<GroupKey, Number> maxMap;
     
-    public ParallelGroupedDoubleDataMaxAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
+    public ParallelGroupedNumberDataMaxAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Maximum");
         maxMap = new HashMap<>();
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Double> element) {
+    protected void handleElement(GroupedDataEntry<Number> element) {
         GroupKey key = element.getKey();
-        Double value = element.getDataEntry();
-        if (!maxMap.containsKey(key) || value > maxMap.get(key)) {
+        Number value = element.getDataEntry();
+        if (!maxMap.containsKey(key) || value.doubleValue() > maxMap.get(key).doubleValue()) {
             maxMap.put(key, value);
         }
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Number> getResult() {
         return maxMap;
     }
 

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

@@ -3,6 +3,7 @@ package com.sap.sse.datamining.impl.components.aggregators;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -15,55 +16,60 @@ import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public class ParallelGroupedDoubleDataMedianAggregationProcessor
-             extends AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> {
+public class ParallelGroupedNumberDataMedianAggregationProcessor
+             extends AbstractParallelGroupedDataStoringAggregationProcessor<Number, Number> {
     
-    private static final AggregationProcessorDefinition<Double, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Median", ParallelGroupedDoubleDataMedianAggregationProcessor.class);
+    private static final AggregationProcessorDefinition<Number, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Median", ParallelGroupedNumberDataMedianAggregationProcessor.class);
     
-    public static AggregationProcessorDefinition<Double, Double> getDefinition() {
+    public static AggregationProcessorDefinition<Number, Number> getDefinition() {
         return DEFINITION;
     }
 
-    private Map<GroupKey, List<Double>> groupedValues;
+    private Map<GroupKey, List<Number>> groupedValues;
 
-    public ParallelGroupedDoubleDataMedianAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
+    public ParallelGroupedNumberDataMedianAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Median");
         groupedValues = new HashMap<>();
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Double> element) {
+    protected void storeElement(GroupedDataEntry<Number> element) {
         GroupKey key = element.getKey();
         if (!groupedValues.containsKey(key)) {
-            groupedValues.put(key, new ArrayList<Double>());
+            groupedValues.put(key, new ArrayList<Number>());
         }
         groupedValues.get(key).add(element.getDataEntry());
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
-        Map<GroupKey, Double> result = new HashMap<>();
-        for (Entry<GroupKey, List<Double>> groupedValuesEntry : groupedValues.entrySet()) {
+    protected Map<GroupKey, Number> aggregateResult() {
+        Map<GroupKey, Number> result = new HashMap<>();
+        for (Entry<GroupKey, List<Number>> groupedValuesEntry : groupedValues.entrySet()) {
             result.put(groupedValuesEntry.getKey(), getMedianOf(groupedValuesEntry.getValue()));
         }
         return result;
     }
 
-    private Double getMedianOf(List<Double> values) {
-        Collections.sort(values);
+    private Double getMedianOf(List<Number> values) {
+        Collections.sort(values, new Comparator<Number>() {
+            @Override
+            public int compare(Number n1, Number n2) {
+                return Double.compare(n1.doubleValue(), n2.doubleValue());
+            }
+        });
         if (listSizeIsEven(values)) {
             int index1 = (values.size() / 2) - 1;
             int index2 = index1 + 1;
-            return (values.get(index1) + values.get(index2)) / 2;
+            return (values.get(index1).doubleValue() + values.get(index2).doubleValue()) / 2;
         } else {
             int index = ((values.size() + 1) / 2) - 1;
-            return values.get(index);
+            return values.get(index).doubleValue();
         }
     }
 
-    private boolean listSizeIsEven(List<Double> values) {
+    private boolean listSizeIsEven(List<?> values) {
         return values.size() % 2 == 0;
     }
 

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

@@ -11,35 +11,35 @@ import com.sap.sse.datamining.impl.components.GroupedDataEntry;
 import com.sap.sse.datamining.impl.components.SimpleAggregationProcessorDefinition;
 import com.sap.sse.datamining.shared.GroupKey;
 
-public class ParallelGroupedDoubleDataMinAggregationProcessor
-                extends AbstractParallelGroupedDataStoringAggregationProcessor<Double, Double> {
+public class ParallelGroupedNumberDataMinAggregationProcessor
+                extends AbstractParallelGroupedDataAggregationProcessor<Number, Number> {
     
-    private static final AggregationProcessorDefinition<Double, Double> DEFINITION =
-            new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Minimum", ParallelGroupedDoubleDataMinAggregationProcessor.class);
+    private static final AggregationProcessorDefinition<Number, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Minimum", ParallelGroupedNumberDataMinAggregationProcessor.class);
     
-    public static AggregationProcessorDefinition<Double, Double> getDefinition() {
+    public static AggregationProcessorDefinition<Number, Number> getDefinition() {
         return DEFINITION;
     }
 
-    private Map<GroupKey, Double> minMap;
+    private Map<GroupKey, Number> minMap;
     
-    public ParallelGroupedDoubleDataMinAggregationProcessor(ExecutorService executor,
-            Collection<Processor<Map<GroupKey, Double>, ?>> resultReceivers) {
+    public ParallelGroupedNumberDataMinAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
         super(executor, resultReceivers, "Minimum");
         minMap = new HashMap<>();
     }
 
     @Override
-    protected void storeElement(GroupedDataEntry<Double> element) {
+    protected void handleElement(GroupedDataEntry<Number> element) {
         GroupKey key = element.getKey();
-        Double value = element.getDataEntry();
-        if (!minMap.containsKey(key) || value < minMap.get(key)) {
+        Number value = element.getDataEntry();
+        if (!minMap.containsKey(key) || value.doubleValue() < minMap.get(key).doubleValue()) {
             minMap.put(key, value);
         }
     }
 
     @Override
-    protected Map<GroupKey, Double> aggregateResult() {
+    protected Map<GroupKey, Number> getResult() {
         return minMap;
     }
 

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

@@ -0,0 +1,47 @@
+package com.sap.sse.datamining.impl.components.aggregators;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+
+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;
+
+public class ParallelGroupedNumberDataSumAggregationProcessor
+             extends AbstractParallelGroupedDataAggregationProcessor<Number, Number> {
+    
+    private static final AggregationProcessorDefinition<Number, Number> DEFINITION =
+            new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Sum", ParallelGroupedNumberDataSumAggregationProcessor.class);
+    
+    public static AggregationProcessorDefinition<Number, Number> getDefinition() {
+        return DEFINITION;
+    }
+
+    private Map<GroupKey, Number> result;
+
+    public ParallelGroupedNumberDataSumAggregationProcessor(ExecutorService executor,
+            Collection<Processor<Map<GroupKey, Number>, ?>> resultReceivers) {
+        super(executor, resultReceivers, "Sum");
+        result = new HashMap<>();
+    }
+    
+    @Override
+    protected void handleElement(GroupedDataEntry<Number> element) {
+        GroupKey key = element.getKey();
+        if (!result.containsKey(key)) {
+            result.put(key, element.getDataEntry());
+        } else {
+            result.put(key, result.get(key).doubleValue() + element.getDataEntry().doubleValue());
+        }
+    }
+    
+    @Override
+    protected Map<GroupKey, Number> getResult() {
+        return result;
+    }
+
+}