java - 使用流式 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 API 收集功能并将上次收集的 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