Parallelized the data grouping

java/com.sap.sailing.datamining.test/src/com/sap/sailing/datamining/test/TestExtractors.java

@@ -1,37 +1,59 @@
 package com.sap.sailing.datamining.test;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.junit.Test;
 
-import com.sap.sailing.datamining.Extractor;
+import com.sap.sailing.datamining.ExtractionWorker;
-import com.sap.sailing.datamining.impl.AbstractExtractor;
+import com.sap.sailing.datamining.WorkReceiver;
-import com.sap.sailing.datamining.shared.Unit;
+import com.sap.sailing.datamining.impl.AbstractExtractionWorker;
+import com.sap.sailing.datamining.shared.GenericGroupKey;
+import com.sap.sailing.datamining.shared.GroupKey;
 
 public class TestExtractors {
 
     @Test
     public void testAbstractExtractor() {
-        Collection<String> data = Arrays.asList("Fu", "Bar", "Blub");
+        ExtractionWorker<String, Integer> lengthExtractor = new LengthExtractor();
-        Extractor<String, Integer> lengthExtractor = new LengthExtractor();
+        DataReceiver receiver = new DataReceiver();
-        Collection<Integer> expectedExtractedData = Arrays.asList(2, 3, 4);
+        lengthExtractor.setReceiver(receiver);
-        assertEquals(expectedExtractedData, lengthExtractor.extract(data));
+        
+        Collection<String> dataEntries = Arrays.asList("Fu", "Bar", "Blub");
+        Map<GroupKey, Collection<String>> data = new HashMap<GroupKey, Collection<String>>();
+        data.put(new GenericGroupKey<Integer>(100), dataEntries);
+        lengthExtractor.setDataToExtractFrom(data);
+        
+
+        Collection<Integer> expectedExtractedDataEntries = Arrays.asList(2, 3, 4);
+        Map<GroupKey, Collection<Integer>> expectedExtractedData = new HashMap<GroupKey, Collection<Integer>>();
+        expectedExtractedData.put(new GenericGroupKey<Integer>(100), expectedExtractedDataEntries);
+        lengthExtractor.run();
+        assertEquals(expectedExtractedData, receiver.result);
     }
     
-    private class LengthExtractor extends AbstractExtractor<String, Integer> {
+    private class LengthExtractor extends AbstractExtractionWorker<String, Integer> {
 
-		public LengthExtractor() {
+        @Override
-			super("", Unit.None, 0);
+        public Integer extract(String dataEntry) {
-		}
+            return dataEntry.length();
+        }
 
-		@Override
+    }
-		public Integer extract(String dataEntry) {
+    
-			return dataEntry.length();
+    private class DataReceiver implements WorkReceiver<Map<GroupKey, Collection<Integer>>> {
-		}
+
-    	
+        public Map<GroupKey, Collection<Integer>> result;
+
+        @Override
+        public void receiveWork(Map<GroupKey, Collection<Integer>> result) {
+            this.result = result;
+        }
+        
     }
 
 }

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

@@ -25,7 +25,7 @@ public final class DataMiningFactory {
         ParallelFilter<DataType> filter = createFilter(queryDefinition);
         
         ParallelGrouper<DataType> grouper = GrouperFactory.createGrouper(queryDefinition, executor);
-        Extractor<DataType, AggregatedType> extractor = ExtractorFactory.createExtractor(queryDefinition.getStatisticType());
+        ParallelExtractor<DataType, AggregatedType> extractor = ExtractorFactory.createExtractor(queryDefinition.getStatisticType(), executor);
         Aggregator<AggregatedType, AggregatedType> aggregator = AggregatorFactory.createAggregator(queryDefinition.getStatisticType(), queryDefinition.getAggregatorType());
         
         return new QueryImpl<DataType, AggregatedType, AggregatedType>(retriever, filter, grouper, extractor, aggregator);

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

@@ -0,0 +1,12 @@
+package com.sap.sailing.datamining;
+
+import java.util.Collection;
+import java.util.Map;
+
+import com.sap.sailing.datamining.shared.GroupKey;
+
+public interface ExtractionWorker<DataType, ExtractedType> extends ComponentWorker<Map<GroupKey, Collection<ExtractedType>>> {
+
+    public void setDataToExtractFrom(Map<GroupKey, Collection<DataType>> data);
+
+}

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

@@ -1,19 +0,0 @@
-package com.sap.sailing.datamining;
-
-import java.util.Collection;
-
-import com.sap.sailing.datamining.shared.Unit;
-
-public interface Extractor<DataType, ExtractedType> {
-
-    public String getSignifier();
-
-    public Unit getUnit();
-
-    public int getValueDecimals();
-
-    public Collection<ExtractedType> extract(Collection<DataType> dataEntry);
-
-    public ExtractedType extract(DataType dataEntry);
-
-}

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

@@ -0,0 +1,19 @@
+package com.sap.sailing.datamining;
+
+import java.util.Collection;
+import java.util.Map;
+
+import com.sap.sailing.datamining.impl.ParallelComponent;
+import com.sap.sailing.datamining.shared.GroupKey;
+import com.sap.sailing.datamining.shared.Unit;
+
+public interface ParallelExtractor<DataType, ExtractedType> extends ParallelComponent<Map<GroupKey, Collection<DataType>>,
+                                                                                      Map<GroupKey, Collection<ExtractedType>>> {
+
+    public Unit getUnit();
+
+    public int getValueDecimals();
+
+    public String getSignifier();
+
+}

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

@@ -12,12 +12,12 @@ import java.util.concurrent.ThreadPoolExecutor;
 import com.sap.sailing.datamining.impl.AbstractParallelComponent;
 import com.sap.sailing.datamining.shared.GroupKey;
 
-public class SimpleParallelGrouper<DataType> extends AbstractParallelComponent<Collection<DataType>, Map<GroupKey, Collection<DataType>>>
+public class PartitioningParallelGrouper<DataType> extends AbstractParallelComponent<Collection<DataType>, Map<GroupKey, Collection<DataType>>>
-                                             implements ParallelGrouper<DataType> {
+                                                   implements ParallelGrouper<DataType> {
 
     private WorkerBuilder<GroupingWorker<DataType>> workerBuilder;
 
-    public SimpleParallelGrouper(WorkerBuilder<GroupingWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
+    public PartitioningParallelGrouper(WorkerBuilder<GroupingWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
         super(executor);
         this.workerBuilder = workerBuilder;
     }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/builders/ExtractionWorkerBuilder.java

@@ -0,0 +1,30 @@
+package com.sap.sailing.datamining.builders;
+
+import com.sap.sailing.datamining.ExtractionWorker;
+import com.sap.sailing.datamining.WorkerBuilder;
+import com.sap.sailing.datamining.impl.SpeedInKnotsExtractor;
+import com.sap.sailing.datamining.impl.trackedLegOfCompetitor.DistanceTraveledExtractor;
+import com.sap.sailing.datamining.shared.Components.StatisticType;
+
+public class ExtractionWorkerBuilder<DataType, ExtractedType> implements WorkerBuilder<ExtractionWorker<DataType, ExtractedType>> {
+
+    private StatisticType statisticType;
+
+    public ExtractionWorkerBuilder(StatisticType statisticType) {
+        this.statisticType = statisticType;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public ExtractionWorker<DataType, ExtractedType> build() {
+        switch (statisticType) {
+        case Speed:
+            return (ExtractionWorker<DataType, ExtractedType>) new SpeedInKnotsExtractor();
+        case Distance_TrackedLegOfCompetitor:
+            return (ExtractionWorker<DataType, ExtractedType>) new DistanceTraveledExtractor();
+        }
+        throw new IllegalArgumentException("Not yet implemented for the given statistic type: "
+                + statisticType.toString());
+    }
+
+}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/factories/DataRetrieverFactory.java

@@ -6,7 +6,7 @@ import com.sap.sailing.datamining.ParallelDataRetriever;
 import com.sap.sailing.datamining.DataRetrievalWorker;
 import com.sap.sailing.datamining.WorkerBuilder;
 import com.sap.sailing.datamining.builders.DataRetrieverWorkerBuilder;
-import com.sap.sailing.datamining.impl.SimpleParallelDataRetriever;
+import com.sap.sailing.datamining.impl.GroupDividingParallelDataRetriever;
 import com.sap.sailing.datamining.shared.DataTypes;
 
 public final class DataRetrieverFactory {
@@ -21,7 +21,7 @@ public final class DataRetrieverFactory {
      */
     public static <DataType> ParallelDataRetriever<DataType> createDataRetriever(DataTypes dataType, ThreadPoolExecutor executor) {
         WorkerBuilder<DataRetrievalWorker<DataType>> workerBuilder = new DataRetrieverWorkerBuilder<DataType>(dataType);
-        return new SimpleParallelDataRetriever<DataType>(workerBuilder, executor);
+        return new GroupDividingParallelDataRetriever<DataType>(workerBuilder, executor);
     }
 
 }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/factories/ExtractorFactory.java

@@ -1,36 +1,22 @@
 package com.sap.sailing.datamining.factories;
 
-import com.sap.sailing.datamining.Extractor;
+import java.util.concurrent.ThreadPoolExecutor;
-import com.sap.sailing.datamining.impl.DataAmountExtractor;
+
-import com.sap.sailing.datamining.impl.SpeedInKnotsExtractor;
+import com.sap.sailing.datamining.ExtractionWorker;
-import com.sap.sailing.datamining.impl.trackedLegOfCompetitor.DistanceTraveledExtractor;
+import com.sap.sailing.datamining.ParallelExtractor;
+import com.sap.sailing.datamining.WorkerBuilder;
+import com.sap.sailing.datamining.builders.ExtractionWorkerBuilder;
+import com.sap.sailing.datamining.impl.GroupDividingParallelExtractor;
 import com.sap.sailing.datamining.shared.Components.StatisticType;
-import com.sap.sailing.domain.base.Moving;
 
 public final class ExtractorFactory {
     
     private ExtractorFactory() { }
     
-    @SuppressWarnings("unchecked")
+    public static <DataType, ExtractedType extends Number> ParallelExtractor<DataType, ExtractedType> createExtractor(StatisticType statisticType, ThreadPoolExecutor executor) {
-    public static <DataType, ExtractedType extends Number> Extractor<DataType, ExtractedType> createExtractor(StatisticType statisticType) {
+        WorkerBuilder<ExtractionWorker<DataType, ExtractedType>> workerBuilder = new ExtractionWorkerBuilder<DataType, ExtractedType>(statisticType);
-        switch (statisticType) {
+        return new GroupDividingParallelExtractor<DataType, ExtractedType>(statisticType.getSignifier(), statisticType.getUnit(), statisticType.getValueDecimals(),
-        case DataAmount:
+                                                                           workerBuilder, executor);
-            return (Extractor<DataType, ExtractedType>) createDataAmountExtractor();
-        case Speed:
-            return (Extractor<DataType, ExtractedType>) createSpeedExtractor();
-        case Distance_TrackedLegOfCompetitor:
-            return (Extractor<DataType, ExtractedType>) new DistanceTraveledExtractor();
-        }
-        throw new IllegalArgumentException("Not yet implemented for the given statistic type: "
-                + statisticType.toString());
-    }
-    
-    public static Extractor<Moving, Double> createSpeedExtractor() {
-        return new SpeedInKnotsExtractor();
-    }
-
-    public static <DataType> Extractor<DataType, Integer> createDataAmountExtractor() {
-        return new DataAmountExtractor<DataType>();
     }
 
 }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/factories/FilterFactory.java

@@ -14,7 +14,7 @@ import com.sap.sailing.datamining.builders.FilterByCriteriaBuilder;
 import com.sap.sailing.datamining.data.impl.NoFilter;
 import com.sap.sailing.datamining.dimensions.DimensionManager;
 import com.sap.sailing.datamining.dimensions.DimensionManagerProvider;
-import com.sap.sailing.datamining.impl.SimpleParallelFilter;
+import com.sap.sailing.datamining.impl.PartitioningParallelFilter;
 import com.sap.sailing.datamining.impl.criterias.AndCompoundFilterCriteria;
 import com.sap.sailing.datamining.impl.criterias.CompoundFilterCriteria;
 import com.sap.sailing.datamining.impl.criterias.DimensionValuesFilterCriteria;
@@ -26,7 +26,7 @@ public final class FilterFactory {
     private FilterFactory() { }
     
     public static <DataType> ParallelFilter<DataType> createParallelFilter(WorkerBuilder<FiltrationWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
-        return new SimpleParallelFilter<DataType>(workerBuilder, executor);
+        return new PartitioningParallelFilter<DataType>(workerBuilder, executor);
     }
 
     public static <DataType> WorkerBuilder<FiltrationWorker<DataType>> createDimensionFilterBuilder(DataTypes dataType, Map<SharedDimension, Iterable<?>> selection) {

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/factories/GrouperFactory.java

@@ -9,7 +9,7 @@ import com.sap.sailing.datamining.BaseBindingProvider;
 import com.sap.sailing.datamining.Dimension;
 import com.sap.sailing.datamining.GroupingWorker;
 import com.sap.sailing.datamining.ParallelGrouper;
-import com.sap.sailing.datamining.SimpleParallelGrouper;
+import com.sap.sailing.datamining.PartitioningParallelGrouper;
 import com.sap.sailing.datamining.WorkerBuilder;
 import com.sap.sailing.datamining.builders.DynamicGrouperBuilder;
 import com.sap.sailing.datamining.builders.GroupByDimensionBuilder;
@@ -26,7 +26,7 @@ public final class GrouperFactory {
 
     public static <DataType> ParallelGrouper<DataType> createGrouper(QueryDefinition queryDefinition, ThreadPoolExecutor executor) {
         WorkerBuilder<GroupingWorker<DataType>> workerBuilder = createGroupingWorkerBuilder(queryDefinition);
-        return new SimpleParallelGrouper<DataType>(workerBuilder, executor);
+        return new PartitioningParallelGrouper<DataType>(workerBuilder, executor);
     }
 
     private static <DataType> WorkerBuilder<GroupingWorker<DataType>> createGroupingWorkerBuilder(QueryDefinition queryDefinition) {

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/AbstractExtractionWorker.java

@@ -0,0 +1,43 @@
+package com.sap.sailing.datamining.impl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import com.sap.sailing.datamining.ExtractionWorker;
+import com.sap.sailing.datamining.shared.GroupKey;
+
+public abstract class AbstractExtractionWorker<DataType, ExtractedType> extends AbstractComponentWorker<Map<GroupKey, Collection<ExtractedType>>>
+                                                                        implements ExtractionWorker<DataType, ExtractedType> {
+
+    private Map<GroupKey, Collection<DataType>> data;
+
+    @Override
+    public void setDataToExtractFrom(Map<GroupKey, Collection<DataType>> data) {
+        this.data = data;
+    }
+
+    @Override
+    protected Map<GroupKey, Collection<ExtractedType>> doWork() {
+        Map<GroupKey, Collection<ExtractedType>> extractedData = new HashMap<GroupKey, Collection<ExtractedType>>();
+        for (Entry<GroupKey, Collection<DataType>> dataEntry : data.entrySet()) {
+            GroupKey key = dataEntry.getKey();
+            if (!extractedData.containsKey(key)) {
+                extractedData.put(key, new ArrayList<ExtractedType>());
+            };
+            
+            for (DataType dateElement : dataEntry.getValue()) {
+                ExtractedType extractedDataElement = extract(dateElement);
+                if (extractedDataElement != null) {
+                    extractedData.get(key).add(extractedDataElement);
+                }
+            }
+        }
+        return extractedData;
+    }
+
+    protected abstract ExtractedType extract(DataType dataEntry);
+
+}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/AbstractExtractor.java

@@ -1,48 +0,0 @@
-package com.sap.sailing.datamining.impl;
-
-import java.util.ArrayList;
-import java.util.Collection;
-
-import com.sap.sailing.datamining.Extractor;
-import com.sap.sailing.datamining.shared.Unit;
-
-public abstract class AbstractExtractor<DataType, ExtractedType> implements Extractor<DataType, ExtractedType> {
-    
-    private String signifier;
-	private Unit unit;
-	private int valueDecimals;
-
-    public AbstractExtractor(String signifier, Unit unit, int valueDecimals) {
-        this.signifier = signifier;
-        this.unit = unit;
-        this.valueDecimals = valueDecimals;
-    }
-    
-    @Override
-    public String getSignifier() {
-        return signifier;
-    }
-    
-    @Override
-    public Unit getUnit() {
-    	return unit;
-    }
-    
-    @Override
-    public int getValueDecimals() {
-    	return valueDecimals;
-    }
-
-    @Override
-    public Collection<ExtractedType> extract(Collection<DataType> data) {
-        Collection<ExtractedType> extractedData = new ArrayList<ExtractedType>();
-        for (DataType dataEntry : data) {
-            ExtractedType extractedDataEntry = extract(dataEntry);
-            if (extractedDataEntry != null) {
-                extractedData.add(extractedDataEntry);
-            }
-        }
-        return extractedData;
-    }
-
-}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/AbstractGroupingWorker.java

@@ -8,8 +8,8 @@ import java.util.Map;
 import com.sap.sailing.datamining.GroupingWorker;
 import com.sap.sailing.datamining.shared.GroupKey;
 
-public abstract class AbstractGrouper<DataType> extends AbstractComponentWorker<Map<GroupKey, Collection<DataType>>>
+public abstract class AbstractGroupingWorker<DataType> extends AbstractComponentWorker<Map<GroupKey, Collection<DataType>>>
-                                                implements GroupingWorker<DataType> {
+                                                       implements GroupingWorker<DataType> {
 
     private Collection<DataType> data;
 

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/DataAmountExtractor.java

@@ -1,26 +0,0 @@
-package com.sap.sailing.datamining.impl;
-
-import java.util.ArrayList;
-import java.util.Collection;
-
-import com.sap.sailing.datamining.shared.Unit;
-
-public class DataAmountExtractor<DataType> extends AbstractExtractor<DataType, Integer> {
-
-    public DataAmountExtractor() {
-        super("amount of fixes", Unit.None, 0);
-    }
-
-    @Override
-    public Collection<Integer> extract(Collection<DataType> data) {
-        Collection<Integer> extractedData = new ArrayList<Integer>();
-        extractedData.add(data.size());
-        return extractedData;
-    }
-
-    @Override
-    public Integer extract(DataType dataEntry) {
-        return 1;
-    }
-
-}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/DynamicGrouper.java

@@ -8,7 +8,7 @@ import com.sap.sailing.datamining.BaseBindingProvider;
 import com.sap.sailing.datamining.shared.GenericGroupKey;
 import com.sap.sailing.datamining.shared.GroupKey;
 
-public class DynamicGrouper<DataType> extends AbstractGrouper<DataType> {
+public class DynamicGrouper<DataType> extends AbstractGroupingWorker<DataType> {
     
     private Script script;
     private Binding binding;

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/GroupByDimension.java

@@ -9,7 +9,7 @@ import com.sap.sailing.datamining.shared.CompoundGroupKey;
 import com.sap.sailing.datamining.shared.GenericGroupKey;
 import com.sap.sailing.datamining.shared.GroupKey;
 
-public class GroupByDimension<DataType, ValueType> extends AbstractGrouper<DataType> {
+public class GroupByDimension<DataType, ValueType> extends AbstractGroupingWorker<DataType> {
     
     private Collection<Dimension<DataType, ValueType>> dimensions;
 

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/GroupDividingParallelDataRetriever.java

@@ -12,16 +12,12 @@ import com.sap.sailing.datamining.WorkerBuilder;
 import com.sap.sailing.domain.leaderboard.LeaderboardGroup;
 import com.sap.sailing.server.RacingEventService;
 
-public class SimpleParallelDataRetriever<DataType> extends AbstractParallelComponent<RacingEventService, Collection<DataType>>
+public class GroupDividingParallelDataRetriever<DataType> extends AbstractParallelComponent<RacingEventService, Collection<DataType>>
-                                                   implements ParallelDataRetriever<DataType> {
+                                                          implements ParallelDataRetriever<DataType> {
 
     private WorkerBuilder<DataRetrievalWorker<DataType>> workerBuilder;
 
-    /**
+    public GroupDividingParallelDataRetriever(WorkerBuilder<DataRetrievalWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
-     * Creates a new parallel working data retriever with the given executer and data retriever as base for the workers.
-     * If a new worker is needed, the worker base will be cloned.
-     */
-    public SimpleParallelDataRetriever(WorkerBuilder<DataRetrievalWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
         super(executor);
         this.workerBuilder = workerBuilder;
     }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/GroupDividingParallelExtractor.java

@@ -0,0 +1,74 @@
+package com.sap.sailing.datamining.impl;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import com.sap.sailing.datamining.ExtractionWorker;
+import com.sap.sailing.datamining.ParallelExtractor;
+import com.sap.sailing.datamining.WorkerBuilder;
+import com.sap.sailing.datamining.shared.GroupKey;
+import com.sap.sailing.datamining.shared.Unit;
+
+public class GroupDividingParallelExtractor<DataType, ExtractedType> extends AbstractParallelComponent<Map<GroupKey, Collection<DataType>>,
+                                                                                                       Map<GroupKey, Collection<ExtractedType>>>
+                                                                     implements ParallelExtractor<DataType, ExtractedType> {
+
+    private String signifier;
+    private Unit unit;
+    private int valueDecimals;
+    
+    private WorkerBuilder<ExtractionWorker<DataType, ExtractedType>> workerBuilder;
+
+    public GroupDividingParallelExtractor(String signifier, Unit unit, int valueDecimals,
+                                          WorkerBuilder<ExtractionWorker<DataType, ExtractedType>> workerBuilder, ThreadPoolExecutor executor) {
+        super(executor);
+        this.signifier = signifier;
+        this.unit = unit;
+        this.valueDecimals = valueDecimals;
+        this.workerBuilder = workerBuilder;
+    }
+
+    @Override
+    protected void setUpWorkersFor(Map<GroupKey, Collection<DataType>> data) {
+        for (Entry<GroupKey, Collection<DataType>> entry : data.entrySet()) {
+            ExtractionWorker<DataType, ExtractedType> worker = workerBuilder.build();
+            worker.setReceiver(this);
+            
+            Map<GroupKey, Collection<DataType>> dataToExtractFrom = new HashMap<GroupKey, Collection<DataType>>();
+            dataToExtractFrom.put(entry.getKey(), entry.getValue());
+            worker.setDataToExtractFrom(dataToExtractFrom);
+            
+            addWorker(worker);
+        }
+    }
+
+    @Override
+    protected Map<GroupKey, Collection<ExtractedType>> finalizeData() {
+        Map<GroupKey, Collection<ExtractedType>> data = new HashMap<GroupKey, Collection<ExtractedType>>();
+        for (Map<GroupKey, Collection<ExtractedType>> result : getResults()) {
+            data.putAll(result);
+        }
+        return Collections.unmodifiableMap(new ConcurrentHashMap<GroupKey, Collection<ExtractedType>>(data));
+    }
+
+    @Override
+    public String getSignifier() {
+        return signifier;
+    }
+
+    @Override
+    public Unit getUnit() {
+        return unit;
+    }
+
+    @Override
+    public int getValueDecimals() {
+        return valueDecimals;
+    }
+
+}

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/PartitioningParallelFilter.java

@@ -11,12 +11,12 @@ import com.sap.sailing.datamining.FiltrationWorker;
 import com.sap.sailing.datamining.ParallelFilter;
 import com.sap.sailing.datamining.WorkerBuilder;
 
-public class SimpleParallelFilter<DataType> extends AbstractParallelComponent<Collection<DataType>, Collection<DataType>>
+public class PartitioningParallelFilter<DataType> extends AbstractParallelComponent<Collection<DataType>, Collection<DataType>>
-                                            implements ParallelFilter<DataType> {
+                                                  implements ParallelFilter<DataType> {
 
     private final WorkerBuilder<FiltrationWorker<DataType>> workerBuilder;
 
-    public SimpleParallelFilter(WorkerBuilder<FiltrationWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
+    public PartitioningParallelFilter(WorkerBuilder<FiltrationWorker<DataType>> workerBuilder, ThreadPoolExecutor executor) {
         super(executor);
         this.workerBuilder = workerBuilder;
     }

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/QueryImpl.java

@@ -6,8 +6,8 @@ import java.util.Map.Entry;
 import java.util.concurrent.ExecutionException;
 
 import com.sap.sailing.datamining.Aggregator;
-import com.sap.sailing.datamining.Extractor;
 import com.sap.sailing.datamining.ParallelDataRetriever;
+import com.sap.sailing.datamining.ParallelExtractor;
 import com.sap.sailing.datamining.ParallelFilter;
 import com.sap.sailing.datamining.ParallelGrouper;
 import com.sap.sailing.datamining.Query;
@@ -22,10 +22,10 @@ public class QueryImpl<DataType, ExtractedType, AggregatedType> implements Query
     private ParallelFilter<DataType> filter;
     private ParallelGrouper<DataType> grouper;
 
-    private Extractor<DataType, ExtractedType> extractor;
+    private ParallelExtractor<DataType, ExtractedType> extractor;
     private Aggregator<ExtractedType, AggregatedType> aggregator;
     
-    public QueryImpl(ParallelDataRetriever<DataType> retriever, ParallelFilter<DataType> filter, ParallelGrouper<DataType> grouper, Extractor<DataType, ExtractedType> extractor, Aggregator<ExtractedType, AggregatedType> aggregator) {
+    public QueryImpl(ParallelDataRetriever<DataType> retriever, ParallelFilter<DataType> filter, ParallelGrouper<DataType> grouper, ParallelExtractor<DataType, ExtractedType> extractor, Aggregator<ExtractedType, AggregatedType> aggregator) {
         this.retriever = retriever;
         this.filter = filter;
         this.grouper = grouper;
@@ -39,12 +39,13 @@ public class QueryImpl<DataType, ExtractedType, AggregatedType> implements Query
         
         Collection<DataType> retrievedData = retriever.start(racingEventService).get();
         Collection<DataType> filteredData = filter.start(retrievedData).get();
+        Map<GroupKey, Collection<DataType>> groupedData = grouper.start(filteredData).get();
+        Map<GroupKey, Collection<ExtractedType>> extractedData = extractor.start(groupedData).get();
+
         QueryResultImpl<AggregatedType> result = new QueryResultImpl<AggregatedType>(retrievedData.size(), filteredData.size(), createResultSignifier(), extractor.getUnit(), extractor.getValueDecimals());
-        Map<GroupKey, Collection<DataType>> groupedFixes = grouper.start(filteredData).get();
+        for (Entry<GroupKey, Collection<ExtractedType>> extractedEntry : extractedData.entrySet()) {
-        for (Entry<GroupKey, Collection<DataType>> groupEntry : groupedFixes.entrySet()) {
+            AggregatedType aggregatedData = aggregator.aggregate(extractedEntry.getValue());
-            Collection<ExtractedType> extractedData = extractor.extract(groupEntry.getValue());
+            result.addResult(extractedEntry.getKey(), aggregatedData);
-            AggregatedType aggregatedData = aggregator.aggregate(extractedData);
-            result.addResult(groupEntry.getKey(), aggregatedData);
         }
         
         final long endTime = System.nanoTime();

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/SpeedInKnotsExtractor.java

@@ -1,13 +1,8 @@
 package com.sap.sailing.datamining.impl;
 
-import com.sap.sailing.datamining.shared.Unit;
 import com.sap.sailing.domain.base.Moving;
 
-public class SpeedInKnotsExtractor extends AbstractExtractor<Moving, Double> {
+public class SpeedInKnotsExtractor extends AbstractExtractionWorker<Moving, Double> {
-
-    public SpeedInKnotsExtractor() {
-        super("speed in knots", Unit.Knots, 2);
-    }
 
     @Override
     public Double extract(Moving dataEntry) {

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/impl/trackedLegOfCompetitor/DistanceTraveledExtractor.java

@@ -1,15 +1,10 @@
 package com.sap.sailing.datamining.impl.trackedLegOfCompetitor;
 
 import com.sap.sailing.datamining.data.TrackedLegOfCompetitorWithContext;
-import com.sap.sailing.datamining.impl.AbstractExtractor;
+import com.sap.sailing.datamining.impl.AbstractExtractionWorker;
-import com.sap.sailing.datamining.shared.Unit;
 import com.sap.sailing.domain.common.Distance;
 
-public class DistanceTraveledExtractor extends AbstractExtractor<TrackedLegOfCompetitorWithContext, Double> {
+public class DistanceTraveledExtractor extends AbstractExtractionWorker<TrackedLegOfCompetitorWithContext, Double> {
-
-    public DistanceTraveledExtractor() {
-        super("distance in meters", Unit.Meters, 2);
-    }
 
     @Override
     public Double extract(TrackedLegOfCompetitorWithContext dataEntry) {

java/com.sap.sailing.datamining/src/com/sap/sailing/datamining/shared/Components.java

@@ -10,23 +10,42 @@ public final class Components {
         Sum, Average, Median
     }
 
+    public static enum ValueType {
+        Integer, Double
+    }
+
     public enum StatisticType {
 
-        DataAmount(ValueType.Integer), Speed(ValueType.Double), Distance_TrackedLegOfCompetitor(ValueType.Double);
+        Speed(ValueType.Double, "speed in knots", Unit.Knots, 2),
-
+        Distance_TrackedLegOfCompetitor(ValueType.Double, "distance in meters", Unit.Meters, 2);
-        public static enum ValueType {
-            Integer, Double
-        }
 
         private ValueType valueType;
+        private String signifier;
+        private Unit unit;
+        private int valueDecimals;
 
-        private StatisticType(ValueType valueType) {
+        private StatisticType(ValueType valueType, String signifier, Unit unit, int valueDecimals) {
             this.valueType = valueType;
+            this.signifier = signifier;
+            this.unit = unit;
+            this.valueDecimals = valueDecimals;
         }
 
         public ValueType getValueType() {
             return valueType;
         }
+        
+        public String getSignifier() {
+            return signifier;
+        }
+        
+        public Unit getUnit() {
+            return unit;
+        }
+        
+        public int getValueDecimals() {
+            return valueDecimals;
+        }
 
     }
 

java/com.sap.sailing.gwt.ui/SailingGWT.launch

@@ -48,7 +48,7 @@
 <stringAttribute key="org.eclipse.jdt.launching.CLASSPATH_PROVIDER" value="com.google.gwt.eclipse.core.moduleClasspathProvider"/>
 <booleanAttribute key="org.eclipse.jdt.launching.DEFAULT_CLASSPATH" value="false"/>
 <stringAttribute key="org.eclipse.jdt.launching.MAIN_TYPE" value="com.google.gwt.dev.DevMode"/>
-<stringAttribute key="org.eclipse.jdt.launching.PROGRAM_ARGUMENTS" value="-war &quot;${project_loc:com.sap.sailing.gwt.ui}&quot; -noserver -remoteUI &quot;${gwt_remote_ui_server_port}:${unique_id}&quot; -logLevel INFO -codeServerPort 9997 -startupUrl /gwt/AdminConsole.html com.sap.sailing.gwt.ui.AdminConsole -startupUrl /gwt/LeaderboardEditing.html com.sap.sailing.gwt.ui.LeaderboardEditing -startupUrl /gwt/UserManagement.html com.sap.sailing.gwt.ui.UserManagement -startupUrl /gwt/Leaderboard.html com.sap.sailing.gwt.ui.Leaderboard -startupUrl /gwt/Spectator.html com.sap.sailing.gwt.ui.Spectator -startupUrl /gwt/RaceBoard.html com.sap.sailing.gwt.ui.RaceBoard -startupUrl /gwt/TvView.html com.sap.sailing.gwt.ui.TvView -startupUrl /gwt/PolarSheets.html com.sap.sailing.gwt.ui.PolarSheets -startupUrl /gwt/RegattaOverview.html com.sap.sailing.gwt.ui.RegattaOverview -startupUrl /gwt/DataMining.html com.sap.sailing.gwt.ui.Simulator"/>
+<stringAttribute key="org.eclipse.jdt.launching.PROGRAM_ARGUMENTS" value="-war &quot;${project_loc:com.sap.sailing.gwt.ui}&quot; -noserver -remoteUI &quot;${gwt_remote_ui_server_port}:${unique_id}&quot; -logLevel INFO -codeServerPort 9997 -startupUrl /gwt/AdminConsole.html com.sap.sailing.gwt.ui.AdminConsole -startupUrl /gwt/LeaderboardEditing.html com.sap.sailing.gwt.ui.LeaderboardEditing -startupUrl /gwt/UserManagement.html com.sap.sailing.gwt.ui.UserManagement -startupUrl /gwt/Leaderboard.html com.sap.sailing.gwt.ui.Leaderboard -startupUrl /gwt/Spectator.html com.sap.sailing.gwt.ui.Spectator -startupUrl /gwt/RaceBoard.html com.sap.sailing.gwt.ui.RaceBoard -startupUrl /gwt/TvView.html com.sap.sailing.gwt.ui.TvView -startupUrl /gwt/PolarSheets.html com.sap.sailing.gwt.ui.PolarSheets -startupUrl /gwt/RegattaOverview.html com.sap.sailing.gwt.ui.RegattaOverview -startupUrl /gwt/DataMining.html com.sap.sailing.gwt.ui.PolarSheets"/>
 <stringAttribute key="org.eclipse.jdt.launching.PROJECT_ATTR" value="com.sap.sailing.gwt.ui"/>
 <stringAttribute key="org.eclipse.jdt.launching.VM_ARGUMENTS" value="-Xmx2048m -XX:MaxPermSize=512m"/>
 </launchConfiguration>

java/com.sap.sailing.gwt.ui/src/main/java/com/sap/sailing/gwt/ui/datamining/selection/SimpleQueryDefinitionProvider.java

@@ -145,11 +145,10 @@ public class SimpleQueryDefinitionProvider extends AbstractQueryDefinitionProvid
                 notifyQueryDefinitionChanged();
             }
         });
-        statisticsProvider.addStatistic(StatisticType.DataAmount, AggregatorType.Sum, DataTypes.GPSFix);
         statisticsProvider.addStatistic(StatisticType.Speed, AggregatorType.Average, DataTypes.GPSFix);
         statisticsProvider.addStatistic(StatisticType.Distance_TrackedLegOfCompetitor, AggregatorType.Sum, DataTypes.TrackedLegOfCompetitor);
         statisticsProvider.addStatistic(StatisticType.Distance_TrackedLegOfCompetitor, AggregatorType.Average, DataTypes.TrackedLegOfCompetitor);
-        statisticsListBox.setValue(statisticsProvider.getStatistic(StatisticType.DataAmount, AggregatorType.Sum, DataTypes.GPSFix), false);
+        statisticsListBox.setValue(statisticsProvider.getStatistic(StatisticType.Speed, AggregatorType.Average, DataTypes.GPSFix), false);
         statisticsListBox.setAcceptableValues(statisticsProvider.getAllStatistics());
         functionsPanel.add(statisticsListBox);