使用流api在列表中搜索值并保持搜索位置(标准化测量数据)

Question

我得到了测量数据的清单。此列表中的项包含时间戳和数据本身。大约每15分钟就有一个条目--但也可能缺少数据点或大抖动。我需要的是建立一个规范化的数据列表，在这里，我每15分钟就有一个条目。作为数据，我可以只进行先前的测量。

输入：

A    B         C     D  E                       F
|----|---------|-----|--|-----------------------|--> t

输出：

|----|----|----|----|----|----|----|----|----|----|--> t
A    B    B    C    C    E    E    E    E    E    F

如何使用java 8中的流以优雅和高效的方式实现这一点？它不可能是一个data.stream().filter([...]).findFirst()，因为可能有很多数据点--从一开始搜索总是太费钱了。我做了同样的测试，输入数据已经对齐到15分钟，这样我就能做到。

public NormalizedData normalizeData(List<MeasurementData> data, Instant t) {
    return data.stream()
        .filter(d -> Objects.equals(d.getTimestamp().getEpochSecond(), t.getEpochSecond()))
        .map(d -> new NormalizedData(t, d))
        .findFirst()
        .orElse(...);
}

对于所有的Instant t来说太慢了。

有什么想法吗？我们应该能够以某种方式将搜索位置存储在流中，并在下一个回合继续。或者完全不同的秘密。如果有与标准流(如StreamEx)兼容的第三方流库的解决方案，这也是一种选择。

Answer 1

下面是一些示例代码，说明如何实现填充缺失的数据点。

下面的许多代码并不是必需的，因为它只是设置数据来显示代码的实际工作方式。

代码所做的是使用Stream收集功能，并将上一次收集的DataPoint与当前的DataPoint进行比较，并在时间戳之间的差值大于15分钟时插入一个新条目。

从测试数据可以看出，C和D以及E和F之间有30分钟的差异，这意味着C和E的数据将被复制。

码

private static final long FIFTEEN_MINS_IN_MILLI_SECONDS = 900_000L;

public static void main(String[] args) {
    //This is just to get some realistic times
    long now = System.currentTimeMillis();
    List<DataPoint> data = getDataPoints(now);

    ArrayList<DataPoint> newDataPoints = data.stream().collect(Collector.of(
            ArrayList<DataPoint>::new,
            (ArrayList<DataPoint> dataPoints, DataPoint nextDataPoint) -> {
                if (!dataPoints.isEmpty()) {
                    addPointIfRequired(dataPoints, nextDataPoint);
                }

                dataPoints.add(nextDataPoint);
            },
            (dataPoints, dataPoints2) -> {
                if (dataPoints.isEmpty()) return dataPoints2;

                if (!dataPoints2.isEmpty()) {
                    addPointIfRequired(dataPoints, dataPoints2.get(0));
                    dataPoints.addAll(dataPoints2);
                }

                return dataPoints;
            }
    ));

    newDataPoints.forEach(System.out::println);
}

private static void addPointIfRequired(ArrayList<DataPoint> dataPoints, DataPoint nextDataPoint) {
    DataPoint previousDataPoint = dataPoints.get(dataPoints.size() - 1);
    long timestampDiff = nextDataPoint.timestamp - previousDataPoint.timestamp;

    if (timestampDiff > FIFTEEN_MINS_IN_MILLI_SECONDS) {
        long fifteenMinIncrement = previousDataPoint.timestamp + FIFTEEN_MINS_IN_MILLI_SECONDS;
        DataPoint newEntry = new DataPoint(previousDataPoint.data, fifteenMinIncrement);
        dataPoints.add(newEntry);
    }
}

private static List<DataPoint> getDataPoints(long now) {
    return Arrays.asList(
            //initial time
            new DataPoint("A", now),
            //15 minute increment
            new DataPoint("B", now + FIFTEEN_MINS_IN_MILLI_SECONDS),
            //15 minute increment
            new DataPoint("C", now + (FIFTEEN_MINS_IN_MILLI_SECONDS * 2)),
            //30 minute increment
            new DataPoint("D", now + (FIFTEEN_MINS_IN_MILLI_SECONDS * 4)),
            //15 minute increment
            new DataPoint("E", now + (FIFTEEN_MINS_IN_MILLI_SECONDS * 5)),
            //30 minute increment
            new DataPoint("F", now + (FIFTEEN_MINS_IN_MILLI_SECONDS * 7))
    );
}

private static class DataPoint {
    private final String data;
    private final long timestamp;

    private DataPoint(String data, long timestamp) {
        this.data = data;
        this.timestamp = timestamp;
    }

    @Override
    public String toString() {
        return data + " " + Instant.ofEpochMilli(timestamp);
    }
}

输出

A 2018-07-11T01:22:45.628Z
B 2018-07-11T01:37:45.628Z
C 2018-07-11T01:52:45.628Z
C 2018-07-11T02:07:45.628Z
D 2018-07-11T02:22:45.628Z
E 2018-07-11T02:37:45.628Z
E 2018-07-11T02:52:45.628Z
F 2018-07-11T03:07:45.628Z

Answer 2

如果我正确理解OP，下面是StreamEx分叉的尝试：

// assume the data is sorted by time
final List<Pair<Integer, String>> data = N.asList(Pair.of(1, "A"), Pair.of(16, "B"), Pair.of(46, "C"),
    Pair.of(60, "D"), Pair.of(76, "E"), Pair.of(151, "F"));
final int startTime = data.get(0).left();
final int interval = 15;

final Map<Integer, Pair<Integer, String>> map = StreamEx.of(data).filter(p -> (p.left() - startTime) % interval == 0).toMap(p -> p.left());

IntStreamEx.rangeClosed(startTime, data.get(data.size() - 1).left(), interval) 
      .forEach(t -> map.computeIfAbsent(t, k -> Pair.of(t, map.get(t - interval).right())));

final List<Pair<Integer, String>> result = StreamEx.of(map).sortedBy(e -> e.getKey()).map(e -> e.getValue()).toList();

System.out.println(result.stream().map(p -> p.right).collect(Collectors.joining("--")));