diff --git a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/Activator.java b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/Activator.java index ae45773df5d..29f8c4a3ac9 100644 --- a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/Activator.java +++ b/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> getAggregationProcessorDefinitions() { HashSet> aggregators = new HashSet<>(); - aggregators.add(ParallelTrueSumAggregationProcessor.getDefinition()); + aggregators.add(ParallelBoolSumAggregationProcessor.getDefinition()); return aggregators; } diff --git a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelTrueSumAggregationProcessor.java b/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelBoolSumAggregationProcessor.java similarity index 75% rename from java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelTrueSumAggregationProcessor.java rename to java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelBoolSumAggregationProcessor.java index d1ec82463c9..1fd4fb39de9 100644 --- a/java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/components/aggregators/ParallelTrueSumAggregationProcessor.java +++ b/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 { +public class ParallelBoolSumAggregationProcessor + extends AbstractParallelGroupedDataAggregationProcessor { private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Boolean.class, Double.class, "Sum", ParallelTrueSumAggregationProcessor.class); + new SimpleAggregationProcessorDefinition<>(Boolean.class, Double.class, "Sum", ParallelBoolSumAggregationProcessor.class); public static AggregationProcessorDefinition getDefinition() { return DEFINITION; @@ -24,14 +24,14 @@ public class ParallelTrueSumAggregationProcessor private Map result; - public ParallelTrueSumAggregationProcessor(ExecutorService executor, + public ParallelBoolSumAggregationProcessor(ExecutorService executor, Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Sum"); result = new HashMap<>(); } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { GroupKey key = element.getKey(); if (!result.containsKey(key)) { result.put(key, 0.0); @@ -43,7 +43,7 @@ public class ParallelTrueSumAggregationProcessor } @Override - protected Map aggregateResult() { + protected Map getResult() { return result; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestStatisticQuery.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestStatisticQuery.java index f7a02fdfe3b..c18c3c0fc38 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/TestStatisticQuery.java +++ b/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 query = server.createQuery(createQueryDefinition()); + Query query = server.createQuery(createQueryDefinition()); assertThat(query.getState(), is(QueryState.NOT_STARTED)); - QueryResult expectedResult = buildExpectedResult(dataSource); + QueryResult 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, Number, Double, Double> createQueryDefinition() { + private StatisticQueryDefinition, 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 sumAggregatorDefinition = ParallelGroupedDoubleDataSumAggregationProcessor.getDefinition(); - ModifiableStatisticQueryDefinition, Number, Double, Double> definition = + AggregationProcessorDefinition sumAggregatorDefinition = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition(); + ModifiableStatisticQueryDefinition, 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 buildExpectedResult(Collection dataSource) { - Map results = new HashMap<>(); + private QueryResult buildExpectedResult(Collection dataSource) { + Map results = new HashMap<>(); results.put(new GenericGroupKey(2), 5.0); results.put(new GenericGroupKey(3), 3.0); results.put(new GenericGroupKey(4), 10.0); - QueryResultImpl result = new QueryResultImpl<>(QueryResultState.NORMAL, Double.class, results, new AdditionalResultDataImpl(dataSource.size() - 2, "Cross Sum (Sum)", Unit.None, "", 0, 0)); + QueryResultImpl 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 result, QueryResult expectedResult) { + private void verifyResult(QueryResult result, QueryResult 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())); diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java index 51bc7aaf86f..8054fd520dc 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestAbstractStoringParallelAggregationProcessor.java +++ b/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> receivers; + private Collection, ?>> receivers; private boolean receiverWasToldToFinish = false; - private Integer receivedElement = null; + private GroupKey groupKey = new GenericGroupKey<>("Key"); + private Map result = null; - private Collection elementStore = new ArrayList<>(); + private Collection> elementStore = new ArrayList<>(); @Before public void initializeReceivers() { - Processor receiver = new NullProcessor(Integer.class, Void.class) { + @SuppressWarnings("unchecked") + Processor, Void> receiver = new NullProcessor, Void>((Class>)(Class) Map.class, Void.class) { @Override - public void processElement(Integer element) { - receivedElement = element; + public void processElement(Map element) { + result = element; } @Override public void finish() throws InterruptedException { @@ -41,43 +48,48 @@ public class TestAbstractStoringParallelAggregationProcessor { @Test public void testAbstractAggregationHandling() throws InterruptedException { - Processor processor = new AbstractParallelStoringAggregationProcessor(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") { + Processor, Map> processor = new AbstractParallelGroupedDataStoringAggregationProcessor(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") { @Override - protected void storeElement(Integer element) { + protected void storeElement(GroupedDataEntry element) { elementStore.add(element); } @Override - protected Integer aggregateResult() { - Integer sum = 0; - for (Integer element : elementStore) { - sum += element; + protected Map aggregateResult() { + Map result = new HashMap<>(); + for (GroupedDataEntry 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 processor, int element) { + private void processElementAndVerifyThatItWasStored(Processor, ?> processor, GroupedDataEntry 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> receivers = new HashSet<>(); - receivers.add(new NullProcessor(Integer.class, Void.class) { + Collection, ?>> receivers = new HashSet<>(); + receivers.add(new NullProcessor, Void>((Class>)(Class) Map.class, Void.class) { @Override - public void processElement(Integer element) { - receivedElement = element; + public void processElement(Map element) { + result = element; } @Override public void finish() throws InterruptedException { @@ -92,32 +104,36 @@ public class TestAbstractStoringParallelAggregationProcessor { } } }); - Processor processor = new AbstractParallelStoringAggregationProcessor(Integer.class, Integer.class, ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") { + Processor, Map> processor = new AbstractParallelGroupedDataStoringAggregationProcessor(ConcurrencyTestsUtil.getExecutor(), receivers, "Sum") { @Override - protected void storeElement(Integer element) { - if (element < 0) { + protected void storeElement(GroupedDataEntry 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 aggregateResult() { + Map result = new HashMap<>(); + for (GroupedDataEntry 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)); } } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelDoubleAggregationProcessors.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelAggregationProcessors.java similarity index 60% rename from java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelDoubleAggregationProcessors.java rename to java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelAggregationProcessors.java index d64ef29a375..1e0eda8bd11 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/impl/components/aggregators/TestParallelDoubleAggregationProcessors.java +++ b/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, ?>> receivers; - private Map receivedAggregations; + private Collection, ?>> receivers; + private Map receivedAggregations; @Test public void testSumAggregationProcessor() throws InterruptedException { - Processor, Map> sumAggregationProcessor = ParallelGroupedDoubleDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); - Collection> elements = createElements(); + Processor, Map> sumAggregationProcessor = ParallelGroupedNumberDataSumAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Collection> elements = createElements(); ConcurrencyTestsUtil.processElements(sumAggregationProcessor, elements); sumAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedSumAggregations(elements); + Map expectedReceivedAggregations = computeExpectedSumAggregations(elements); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedSumAggregations(Collection> elements) { - Map expectedSumAggregations = new HashMap<>(); - for (GroupedDataEntry element : elements) { + private Map computeExpectedSumAggregations(Collection> elements) { + Map expectedSumAggregations = new HashMap<>(); + for (GroupedDataEntry 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, Map> averageAggregationProcessor = ParallelGroupedDoubleDataAverageAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); - Collection> elements = createElements(); + Processor, Map> averageAggregationProcessor = ParallelGroupedNumberDataAverageAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Collection> elements = createElements(); ConcurrencyTestsUtil.processElements(averageAggregationProcessor, elements); averageAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedAverageAggregations(elements); + Map expectedReceivedAggregations = computeExpectedAverageAggregations(elements); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedAverageAggregations(Collection> elements) { - Map result = new HashMap<>(); + private Map computeExpectedAverageAggregations(Collection> elements) { + Map result = new HashMap<>(); - Map sumAggregations = computeExpectedSumAggregations(elements); + Map sumAggregations = computeExpectedSumAggregations(elements); Map elementAmountPerKey = countElementAmountPerKey(elements); - for (Entry sumAggregationEntry : sumAggregations.entrySet()) { + for (Entry 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 countElementAmountPerKey(Collection> elements) { + private Map countElementAmountPerKey(Collection> elements) { Map elementAmountPerKey = new HashMap<>(); - for (GroupedDataEntry 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, Map> medianAggregationProcessor = ParallelGroupedDoubleDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); - Collection> elements = createElements(); + Processor, Map> medianAggregationProcessor = ParallelGroupedNumberDataMedianAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Collection> elements = createElements(); ConcurrencyTestsUtil.processElements(medianAggregationProcessor, elements); medianAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedMedianAggregations(); + Map expectedReceivedAggregations = computeExpectedMedianAggregations(); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedMedianAggregations() { - Map result = new HashMap<>(); + private Map computeExpectedMedianAggregations() { + Map result = new HashMap<>(); result.put(new GenericGroupKey(1), 7.0); result.put(new GenericGroupKey(2), 6.0); result.put(new GenericGroupKey(3), 5.0); @@ -102,17 +102,17 @@ public class TestParallelDoubleAggregationProcessors { @Test public void testMaxAggregationProcessor() throws InterruptedException { - Processor, Map> maxAggregationProcessor = ParallelGroupedDoubleDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); - Collection> elements = createElements(); + Processor, Map> maxAggregationProcessor = ParallelGroupedNumberDataMaxAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Collection> elements = createElements(); ConcurrencyTestsUtil.processElements(maxAggregationProcessor, elements); maxAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedMaxAggregations(); + Map expectedReceivedAggregations = computeExpectedMaxAggregations(); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedMaxAggregations() { - Map result = new HashMap<>(); + private Map computeExpectedMaxAggregations() { + Map result = new HashMap<>(); result.put(new GenericGroupKey(1), 10.0); result.put(new GenericGroupKey(2), 7.0); result.put(new GenericGroupKey(3), 5.0); @@ -121,17 +121,17 @@ public class TestParallelDoubleAggregationProcessors { @Test public void testMinAggregationProcessor() throws InterruptedException { - Processor, Map> minAggregationProcessor = ParallelGroupedDoubleDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); - Collection> elements = createElements(); + Processor, Map> minAggregationProcessor = ParallelGroupedNumberDataMinAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Collection> elements = createElements(); ConcurrencyTestsUtil.processElements(minAggregationProcessor, elements); minAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedMinAggregations(); + Map expectedReceivedAggregations = computeExpectedMinAggregations(); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedMinAggregations() { - Map result = new HashMap<>(); + private Map computeExpectedMinAggregations() { + Map result = new HashMap<>(); result.put(new GenericGroupKey(1), 5.0); result.put(new GenericGroupKey(2), 3.0); result.put(new GenericGroupKey(3), 5.0); @@ -140,42 +140,42 @@ public class TestParallelDoubleAggregationProcessors { @Test public void testCountAggregationProcessor() throws InterruptedException { - Processor, Map> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); + Processor, Map> countAggregationProcessor = ParallelGroupedDataCountAggregationProcessor.getDefinition().construct(ConcurrencyTestsUtil.getExecutor(), receivers); @SuppressWarnings("unchecked") Collection> elements = (Collection>)(Collection) createElements(); ConcurrencyTestsUtil.processElements(countAggregationProcessor, elements); countAggregationProcessor.finish(); - Map expectedReceivedAggregations = computeExpectedCountAggregations(); + Map expectedReceivedAggregations = computeExpectedCountAggregations(); ConcurrencyTestsUtil.verifyResultData(receivedAggregations, expectedReceivedAggregations); } - private Map computeExpectedCountAggregations() { - Map result = new HashMap<>(); + private Map computeExpectedCountAggregations() { + Map result = new HashMap<>(); result.put(new GenericGroupKey(1), 3.0); result.put(new GenericGroupKey(2), 4.0); result.put(new GenericGroupKey(3), 3.0); return result; } - private Collection> createElements() { - Collection> elements = new ArrayList<>(); + private Collection> createElements() { + Collection> elements = new ArrayList<>(); GroupKey firstGroupKey = new GenericGroupKey(1); - elements.add(new GroupedDataEntry(firstGroupKey, 5.0)); - elements.add(new GroupedDataEntry(firstGroupKey, 10.0)); - elements.add(new GroupedDataEntry(firstGroupKey, 7.0)); + elements.add(new GroupedDataEntry(firstGroupKey, 5.0)); + elements.add(new GroupedDataEntry(firstGroupKey, 10.0)); + elements.add(new GroupedDataEntry(firstGroupKey, 7.0)); GroupKey secondGroupKey = new GenericGroupKey(2); - elements.add(new GroupedDataEntry(secondGroupKey, 5.0)); - elements.add(new GroupedDataEntry(secondGroupKey, 3.0)); - elements.add(new GroupedDataEntry(secondGroupKey, 7.0)); - elements.add(new GroupedDataEntry(secondGroupKey, 7.0)); + elements.add(new GroupedDataEntry(secondGroupKey, 5.0)); + elements.add(new GroupedDataEntry(secondGroupKey, 3.0)); + elements.add(new GroupedDataEntry(secondGroupKey, 7.0)); + elements.add(new GroupedDataEntry(secondGroupKey, 7.0)); GroupKey thirdGroupKey = new GenericGroupKey(3); - elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); - elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); - elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); + elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); + elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); + elements.add(new GroupedDataEntry(thirdGroupKey, 5.0)); return elements; } @@ -183,9 +183,9 @@ public class TestParallelDoubleAggregationProcessors { @Before public void initializeResultReceivers() { @SuppressWarnings("unchecked") - Processor, Void> receiver = new NullProcessor, Void>((Class>)(Class) Map.class, Void.class) { + Processor, Void> receiver = new NullProcessor, Void>((Class>)(Class) Map.class, Void.class) { @Override - public void processElement(Map element) { + public void processElement(Map element) { receivedAggregations = element; } }; diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAggregationProcessorDummyWithInvalidConstructor.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAggregationProcessorDummyWithInvalidConstructor.java index a5799631dfc..6072bbdc4ba 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAggregationProcessorDummyWithInvalidConstructor.java +++ b/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 { + AbstractParallelGroupedDataAggregationProcessor { public TestAggregationProcessorDummyWithInvalidConstructor(ExecutorService executor, Collection, ?>> resultReceivers, Object constructorInvalidator) { @@ -18,11 +18,11 @@ public class TestAggregationProcessorDummyWithInvalidConstructor extends } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { } @Override - protected Map aggregateResult() { + protected Map getResult() { return null; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAnyDataTypeAggregationProcessorDummy.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAnyDataTypeAggregationProcessorDummy.java index d46e69651b6..6f763c5be49 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestAnyDataTypeAggregationProcessorDummy.java +++ b/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 { + AbstractParallelGroupedDataAggregationProcessor { public TestAnyDataTypeAggregationProcessorDummy(ExecutorService executor, Collection, ?>> resultReceivers) { @@ -18,11 +18,11 @@ public class TestAnyDataTypeAggregationProcessorDummy extends } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { } @Override - protected Map aggregateResult() { + protected Map getResult() { return null; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestCompetitorAggregationProcessorDummy.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestCompetitorAggregationProcessorDummy.java index fbec9447088..c5feb1d991f 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestCompetitorAggregationProcessorDummy.java +++ b/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 { + AbstractParallelGroupedDataAggregationProcessor { public TestCompetitorAggregationProcessorDummy(ExecutorService executor, Collection, ?>> resultReceivers) { @@ -19,11 +19,11 @@ public class TestCompetitorAggregationProcessorDummy extends } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { } @Override - protected Map aggregateResult() { + protected Map getResult() { return null; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestNamedAggregationProcessorDummy.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestNamedAggregationProcessorDummy.java index ab5cd9f466f..7cebdad8fb9 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestNamedAggregationProcessorDummy.java +++ b/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 { + AbstractParallelGroupedDataAggregationProcessor { public TestNamedAggregationProcessorDummy(ExecutorService executor, Collection, ?>> resultReceivers) { @@ -19,11 +19,11 @@ public class TestNamedAggregationProcessorDummy extends } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { } @Override - protected Map aggregateResult() { + protected Map getResult() { return null; } diff --git a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestRaceAggregationProcessorDummy.java b/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestRaceAggregationProcessorDummy.java index 1abfa703a1b..f8d75b1b88a 100644 --- a/java/com.sap.sse.datamining.test/src/com/sap/sse/datamining/test/util/components/aggregators/TestRaceAggregationProcessorDummy.java +++ b/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 { + AbstractParallelGroupedDataAggregationProcessor { public TestRaceAggregationProcessorDummy(ExecutorService executor, Collection, ?>> resultReceivers) { @@ -19,11 +19,11 @@ public class TestRaceAggregationProcessorDummy extends } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry element) { } @Override - protected Map aggregateResult() { + protected Map getResult() { return null; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/AggregationProcessorDefinition.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/AggregationProcessorDefinition.java index 99be8f22b4b..ba032b6290c 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/components/AggregationProcessorDefinition.java +++ b/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 { @@ -15,7 +15,7 @@ public interface AggregationProcessorDefinition { public String getAggregationNameMessageKey(); - public Class> getAggregationProcessor(); + public Class> getAggregationProcessor(); public Processor, Map> construct(ExecutorService executor, Collection, ?>> resultReceivers); } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/DataMiningFrameworkActivator.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/DataMiningFrameworkActivator.java index 31b0c2611f7..46c07276583 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/DataMiningFrameworkActivator.java +++ b/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> getDefaultAggregationProcessors() { Collection> 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; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/SimpleAggregationProcessorDefinition.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/SimpleAggregationProcessorDefinition.java index ebd75806d44..45e1f14c2d2 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/SimpleAggregationProcessorDefinition.java +++ b/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 implements @@ -18,14 +18,14 @@ public class SimpleAggregationProcessorDefinition private final Class extractedType; private final Class aggregatedType; private final String aggregationNameMessageKey; - private final Class> aggregationProcessor; - private final Constructor> constructor; + private final Class> aggregationProcessor; + private final Constructor> constructor; public SimpleAggregationProcessorDefinition( Class extractedType, Class aggregatedType, String aggregationNameMessageKey, - Class> aggregationProcessor) { + Class> aggregationProcessor) { this.extractedType = extractedType; this.aggregatedType = aggregatedType; this.aggregationNameMessageKey = aggregationNameMessageKey; @@ -33,9 +33,9 @@ public class SimpleAggregationProcessorDefinition constructor = ensureValidConstructor(aggregationProcessor); } - private Constructor> ensureValidConstructor(Class> aggregationProcessor) { + private Constructor> ensureValidConstructor(Class> aggregationProcessor) { try { - Constructor> constructor = aggregationProcessor.getConstructor(ExecutorService.class, Collection.class); + Constructor> 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 } @Override - public Class> getAggregationProcessor() { + public Class> getAggregationProcessor() { return aggregationProcessor; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java similarity index 58% rename from java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java rename to java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelAggregationProcessor.java index 1cde8e1bc04..f8856b641ba 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelStoringAggregationProcessor.java +++ b/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 +public abstract class AbstractParallelAggregationProcessor extends AbstractParallelProcessor { - private final Lock storeLock; private final String aggregationNameMessageKey; + private final Lock writeLock; - public AbstractParallelStoringAggregationProcessor(Class inputType, - Class resultType, - ExecutorService executor, - Collection> resultReceivers, - String aggregationNameMessageKey) { + public AbstractParallelAggregationProcessor(Class inputType, + Class resultType, + ExecutorService executor, + Collection> 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(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) { diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelGroupedDataAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelGroupedDataAggregationProcessor.java new file mode 100644 index 00000000000..9c44671636d --- /dev/null +++ b/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 + extends AbstractParallelAggregationProcessor, Map> { + + @SuppressWarnings("unchecked") + public AbstractParallelGroupedDataAggregationProcessor(ExecutorService executor, Collection, ?>> resultReceivers, String aggregationNameMessageKey) { + super((Class>)(Class) GroupedDataEntry.class, (Class>)(Class) Map.class, + executor, resultReceivers, aggregationNameMessageKey); + } + +} diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelGroupedDataStoringAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelGroupedDataStoringAggregationProcessor.java index 4f5df432a28..43698257c1f 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/AbstractParallelGroupedDataStoringAggregationProcessor.java +++ b/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 - extends AbstractParallelStoringAggregationProcessor, Map> { +public abstract class AbstractParallelGroupedDataStoringAggregationProcessor + extends AbstractParallelGroupedDataAggregationProcessor { - @SuppressWarnings("unchecked") - public AbstractParallelGroupedDataStoringAggregationProcessor(ExecutorService executor, Collection, ?>> resultReceivers, String aggregationNameMessageKey) { - super((Class>)(Class) GroupedDataEntry.class, (Class>)(Class) Map.class, - executor, resultReceivers, aggregationNameMessageKey); + public AbstractParallelGroupedDataStoringAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers, + String aggregationNameMessageKey) { + super(executor, resultReceivers, aggregationNameMessageKey); } + @Override + protected void handleElement(GroupedDataEntry 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 element); + + @Override + protected Map getResult() { + return aggregateResult(); + } + + protected abstract Map aggregateResult(); + } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDataCountAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDataCountAggregationProcessor.java index fb63bc7ec3d..1777e277543 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDataCountAggregationProcessor.java +++ b/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 { + extends AbstractParallelGroupedDataAggregationProcessor { - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Object.class, Double.class, "Count", ParallelGroupedDataCountAggregationProcessor.class); + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Object.class, Number.class, "Count", ParallelGroupedDataCountAggregationProcessor.class); - public static AggregationProcessorDefinition getDefinition() { + public static AggregationProcessorDefinition getDefinition() { return DEFINITION; } - private Map countMap; + private Map countMap; public ParallelGroupedDataCountAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { + Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Count"); countMap = new HashMap<>(); } - + @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry 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 aggregateResult() { + protected Map getResult() { return countMap; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataSumAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataSumAggregationProcessor.java deleted file mode 100644 index 42e429a4549..00000000000 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataSumAggregationProcessor.java +++ /dev/null @@ -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 { - - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Sum", ParallelGroupedDoubleDataSumAggregationProcessor.class); - - public static AggregationProcessorDefinition getDefinition() { - return DEFINITION; - } - - private Map, Integer> elementAmountMap; - - public ParallelGroupedDoubleDataSumAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { - super(executor, resultReceivers, "Sum"); - elementAmountMap = new HashMap<>(); - } - - @Override - protected void storeElement(GroupedDataEntry element) { - if (!elementAmountMap.containsKey(element)) { - elementAmountMap.put(element, 0); - } - Integer currentAmount = elementAmountMap.get(element); - elementAmountMap.put(element, currentAmount + 1); - } - - @Override - protected Map aggregateResult() { - Map result = new HashMap<>(); - for (Entry, 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); - } - -} diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataAverageAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataAverageAggregationProcessor.java similarity index 54% rename from java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataAverageAggregationProcessor.java rename to java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataAverageAggregationProcessor.java index 1cc5a8df412..9ddf94e65e2 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataAverageAggregationProcessor.java +++ b/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 { +public class ParallelGroupedNumberDataAverageAggregationProcessor + extends AbstractParallelGroupedDataStoringAggregationProcessor { - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Average", ParallelGroupedDoubleDataAverageAggregationProcessor.class); + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Average", ParallelGroupedNumberDataAverageAggregationProcessor.class); - public static AggregationProcessorDefinition getDefinition() { + public static AggregationProcessorDefinition getDefinition() { return DEFINITION; } - private final AbstractParallelGroupedDataStoringAggregationProcessor sumAggregationProcessor; + private final ParallelGroupedNumberDataSumAggregationProcessor sumAggregationProcessor; private final Map elementAmountPerKey; - public ParallelGroupedDoubleDataAverageAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { + public ParallelGroupedNumberDataAverageAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Average"); elementAmountPerKey = new HashMap<>(); - sumAggregationProcessor = new ParallelGroupedDoubleDataSumAggregationProcessor(executor, resultReceivers); + sumAggregationProcessor = new ParallelGroupedNumberDataSumAggregationProcessor(executor, resultReceivers); } @Override - protected void storeElement(GroupedDataEntry element) { + protected void storeElement(GroupedDataEntry element) { incrementElementAmount(element); - sumAggregationProcessor.storeElement(element); + sumAggregationProcessor.handleElement(element); } - private void incrementElementAmount(GroupedDataEntry element) { + private void incrementElementAmount(GroupedDataEntry element) { GroupKey key = element.getKey(); if (!elementAmountPerKey.containsKey(key)) { elementAmountPerKey.put(key, 0); @@ -48,12 +48,12 @@ public class ParallelGroupedDoubleDataAverageAggregationProcessor } @Override - protected Map aggregateResult() { - Map result = new HashMap<>(); - Map sumAggregation = sumAggregationProcessor.aggregateResult(); - for (Entry sumAggregationEntry : sumAggregation.entrySet()) { + protected Map aggregateResult() { + Map result = new HashMap<>(); + Map sumAggregation = sumAggregationProcessor.getResult(); + for (Entry 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; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMaxAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMaxAggregationProcessor.java similarity index 50% rename from java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMaxAggregationProcessor.java rename to java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMaxAggregationProcessor.java index 79cf5865666..fab472527fb 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMaxAggregationProcessor.java +++ b/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 { +public class ParallelGroupedNumberDataMaxAggregationProcessor + extends AbstractParallelGroupedDataAggregationProcessor { - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Maximum", ParallelGroupedDoubleDataMaxAggregationProcessor.class); + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Maximum", ParallelGroupedNumberDataMaxAggregationProcessor.class); - public static AggregationProcessorDefinition getDefinition() { + public static AggregationProcessorDefinition getDefinition() { return DEFINITION; } - private Map maxMap; + private Map maxMap; - public ParallelGroupedDoubleDataMaxAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { + public ParallelGroupedNumberDataMaxAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Maximum"); maxMap = new HashMap<>(); } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry 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 aggregateResult() { + protected Map getResult() { return maxMap; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMedianAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMedianAggregationProcessor.java similarity index 52% rename from java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMedianAggregationProcessor.java rename to java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMedianAggregationProcessor.java index 9bd93da8999..26da6a4f8e3 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMedianAggregationProcessor.java +++ b/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 { +public class ParallelGroupedNumberDataMedianAggregationProcessor + extends AbstractParallelGroupedDataStoringAggregationProcessor { - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Median", ParallelGroupedDoubleDataMedianAggregationProcessor.class); + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Median", ParallelGroupedNumberDataMedianAggregationProcessor.class); - public static AggregationProcessorDefinition getDefinition() { + public static AggregationProcessorDefinition getDefinition() { return DEFINITION; } - private Map> groupedValues; + private Map> groupedValues; - public ParallelGroupedDoubleDataMedianAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { + public ParallelGroupedNumberDataMedianAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Median"); groupedValues = new HashMap<>(); } @Override - protected void storeElement(GroupedDataEntry element) { + protected void storeElement(GroupedDataEntry element) { GroupKey key = element.getKey(); if (!groupedValues.containsKey(key)) { - groupedValues.put(key, new ArrayList()); + groupedValues.put(key, new ArrayList()); } groupedValues.get(key).add(element.getDataEntry()); } @Override - protected Map aggregateResult() { - Map result = new HashMap<>(); - for (Entry> groupedValuesEntry : groupedValues.entrySet()) { + protected Map aggregateResult() { + Map result = new HashMap<>(); + for (Entry> groupedValuesEntry : groupedValues.entrySet()) { result.put(groupedValuesEntry.getKey(), getMedianOf(groupedValuesEntry.getValue())); } return result; } - private Double getMedianOf(List values) { - Collections.sort(values); + private Double getMedianOf(List values) { + Collections.sort(values, new Comparator() { + @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 values) { + private boolean listSizeIsEven(List values) { return values.size() % 2 == 0; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMinAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMinAggregationProcessor.java similarity index 50% rename from java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMinAggregationProcessor.java rename to java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataMinAggregationProcessor.java index 152536b99db..7f23c08d727 100644 --- a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedDoubleDataMinAggregationProcessor.java +++ b/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 { +public class ParallelGroupedNumberDataMinAggregationProcessor + extends AbstractParallelGroupedDataAggregationProcessor { - private static final AggregationProcessorDefinition DEFINITION = - new SimpleAggregationProcessorDefinition<>(Double.class, Double.class, "Minimum", ParallelGroupedDoubleDataMinAggregationProcessor.class); + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Minimum", ParallelGroupedNumberDataMinAggregationProcessor.class); - public static AggregationProcessorDefinition getDefinition() { + public static AggregationProcessorDefinition getDefinition() { return DEFINITION; } - private Map minMap; + private Map minMap; - public ParallelGroupedDoubleDataMinAggregationProcessor(ExecutorService executor, - Collection, ?>> resultReceivers) { + public ParallelGroupedNumberDataMinAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers) { super(executor, resultReceivers, "Minimum"); minMap = new HashMap<>(); } @Override - protected void storeElement(GroupedDataEntry element) { + protected void handleElement(GroupedDataEntry 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 aggregateResult() { + protected Map getResult() { return minMap; } diff --git a/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataSumAggregationProcessor.java b/java/com.sap.sse.datamining/src/com/sap/sse/datamining/impl/components/aggregators/ParallelGroupedNumberDataSumAggregationProcessor.java new file mode 100644 index 00000000000..f16105dc54d --- /dev/null +++ b/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 { + + private static final AggregationProcessorDefinition DEFINITION = + new SimpleAggregationProcessorDefinition<>(Number.class, Number.class, "Sum", ParallelGroupedNumberDataSumAggregationProcessor.class); + + public static AggregationProcessorDefinition getDefinition() { + return DEFINITION; + } + + private Map result; + + public ParallelGroupedNumberDataSumAggregationProcessor(ExecutorService executor, + Collection, ?>> resultReceivers) { + super(executor, resultReceivers, "Sum"); + result = new HashMap<>(); + } + + @Override + protected void handleElement(GroupedDataEntry 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 getResult() { + return result; + } + +}