Accumulating value of objects when carrying the same timestamp

user_5476187 0
Viewed 87

I am currently stuck on this:

I have datapoints that carry a value and a timestamp as a Long (epoch seconds):

public class MyDataPoint(){
    private Float value;
    private Long timestamp;

   //constructor, getters and setters here
}

I have lists that are bound to different sources where these datapoints are coming from.

public class MySource(){
    private Interger sourceId;
    private List<MyDataPoint> dataPointList;

    //constructor, getters and setters here
}

Now I want to accumulate these datapoints in a new list: each datapoint with the same timestamp should be accumulated in a new datapoint with the sum of the value of each datapoint that carries the same timestamp.

So for instance I have 3 datapoints with the same timestamp, I want to create one datapoint with the timestamp, and the sum of the three values.

However, these datapoints have not started or ended recording at the same time. And for one timestamp maybe only one datapoint exists.

For now I have stuffed all of the datapoints into one list, thinking I could use streams to achieve my goal, but I can't figure it out. Maybe this is the wrong way anyway because I can't see how to use filters or maps to do this.

I have thought about using Optionals since for one timestamp maybe only one exists, but there is no obvious answer for me.

Anyone able to help me out?

java-8 java
3 Answers

I am guessing that you are trying to grouping the value you in the list, then convert it to new list using stream. What i suggest is using Collectors.groupingBy and Collectors.summingInt to convert your List to a Map<Long,Double> first - which holding your timestamp as key and Double as sum of all value that has same timestamp. After this you can convert this map back to the new list.

Not tested yet but to convert your List to Map<Long, Double> should be something like:

dataPointList.stream().collect(Collectors.groupingBy(d -> d.timestamp, Collectors.summingDouble(d -> d.value))); //you can using method reference for better readability
user_114908080

I think the "big picture" solution it's quite easy even if I can predict some multithread issues to complicate all.

In pure Java, you need simply a Map:

Map<Long,List<MyDataPoint>> dataPoints = new HashMap<>();

just use Timestamp as KEY.

For the sake of OOP, Let's create a class like DataPointCollector

   public class DataPointCollector {
    private Map<Long,List<MyDataPoint>> dataPoints = new HashMap<>();
    
    }

To add element, create a method in DataPointCollector like:

public void addDataPoint(MyDataPoint dp){

  if (dataPoints.get(dp.getTimestamp()) == null){
    dataPoints.put(dp.getTimestamp(), new ArrayList<MyDataPoint>());
}
  
dataPoints.get(dp.getTimestamp()).add(dp);
}

This solve most of your theorical problems. To get the sum, just iterate over the List and sum the values.

If you need a realtime sum, just wrap the List in another object that has totalValue and List<MyDataPoint> as fields and update totalValue on each invokation of addDataPoint(...).

About streams: streams depends by use cases, if in a certain time you have all the DataPoints you need, of course you can use Streams to do things... however streams are often expensive for common cases and I think it's better to focus on an easy solution and then make it cool with streams only if needed

user_33469140

Following assumes your DataPoint is immutable (you cannot use the same instance to accumulate into) so uses an intermediate Map.

Collection<DataPoint> summary = sources.stream()
      .flatMap(source -> source.dataPointList.stream()) // smush sources into a single stream of points
      .collect(groupingBy(p -> p.timestamp, summingDouble(p -> (double)p.value))) // Collect points into Map<Long, Double>
      .entrySet().stream() // New stream, the entries of the Map
      .map(e -> new MyDataPoint(e.getKey(), e.getValue()))
      .collect(toList());

Another solution avoids the potentially large intermediate Map by collecting directly into a DataPoint.

public static DataPoint combine(DataPoint left, DataPoint right) {
  return new DataPoint(left.timestamp, left.value + right.value);  // return new if immutable or increase left if not
}

Collection<DataPoint> summary = sources.stream()
      .flatMap(source -> source.dataPointList.stream()) // smush into a single stream of points
      .collect(groupingBy(p -> p.timestamp, reducing(DataPoint.ZERO, DataPoint::combine))) // Collect all values into Map<Long, DataPoint>
      .values();

This can be upgraded to parallelStream() if DataPoint is threadsafe etc

user_6283180
Related