基于列串列對資料進行分組和聚合

我有一個包含多個 JSONObjects 的 JSONArray，每個 JSON 物件代表一行資料。（就像一個 SQL 行）

例子：

[{
    "col1": "c1",
    "col2": "r1",
    "col3": 12121
}, {
    "col1": "c1",
    "col2": "r1",
    "col3": 1321
}, {
    "col1": "c1",
    "col2": "r2",
    "col3": 4342
}, {
    "col1": "c1",
    "col2": "r2",
    "col3": 4532
}]

包含必須發生 group by 的列的串列：

例子：

["col1","col2"]

最后，必須應用的聚合、MIN、MAX、SUM，AVG以及必須應用聚合的列：

預期輸出：合計為 SUM

[{
    "col1": "c1",
    "col2": "r1",
    "col3": 13442
},{
    "col1": "c1",
    "col2": "r2",
    "col3": 8874
}]

到目前為止我嘗試過的：

我想將當前與以前的列串列進行比較，每當我看到值發生變化時，我都會對其進行匯總。但是這種方法看起來效率太低了。我正在考慮使用 Java Streams，但我很不擅長。任何幫助，將不勝感激。

 if (agg.equalsIgnoreCase("MIN")) {
        Number min = data.getJSONObject(0).getNumber(column);
        for (int i = 0; i < data.length(); i  ) {
            JSONObject jsonObject = data.getJSONObject(i);
            if (i > 1) {
            }
        }
    }

uj5u.com熱心網友回復：

根據您將要處理的資料量，一種不依賴于流的簡單方法是使用Map. 散列聚合列值以生成映射鍵，并根據聚合列的值更新映射值。

在這里，我創建了一個Operation可以為每個操作（總和、最大值、最小值等）實作的介面。

例如

interface Operation {
    Long update(Long currentAggregate, int nextValue);
}

class Sum implements Operation {
    @Override
    public Long update(Long currentAggregate, int nextValue) {
        return currentAggregate   nextValue;
    }
}

JSONArray aggregate(JSONArray array, String[] columns, String aggregateColumn, Operation op) {
    Map<String, Long> aggregates = new HashMap<>();
    for (int i = 0; i < array.size();   i) {
        JSONObject obj = array.getJsonObject(i);
        String key = getKey(obj, columns);
        Long current = aggregates.get(key);
        aggregates.put(key, op.update(current, obj.getInt(aggregateColumn)));
    }
    // Then split the map key back out to columns values (or use a more sophisticated 
    // object in place of 'aggregates' that also stores the column values explicitly) and 
    // return a JSONArray with values for the 'aggregateColumn' taken from 'aggregates'.
    // ...
}

String getKey(JSONObject obj, String[] columns) {
    // Assumes no column names include "_".
    StringBuilder builder = new StringBuilder();
    for (int i = 0; i < columns.length;   i)
        builder.append(obj.getString(columns[i])).append("_");
    return builder.toString();
}

uj5u.com熱心網友回復：

您必須先分解問題，然后再意識到這并不難，尤其是當有可以滿足您需求的工具時，您甚至不需要自己實作某些東西。流在這里完全是一個不錯的選擇，因為 Java 8 Streams API 允許按鍵對流元素進行分組，并將組作為下游進行處理，例如，聚合操作。

假設您有一個生成巨大資料集的 JSON 源：對于您的示例，它仍然可以表示為Stream<JSONObject>. 我使用您的檔案以流式方式讀取它，生成可供分析的資料流（我打賭我的拆分器實作并不完美，但它似乎有效）：

public static <T> Stream<T> asStream(final JSONTokener jsonTokener) {
    return StreamSupport.stream(new Spliterators.AbstractSpliterator<>(Long.MAX_VALUE, Spliterator.IMMUTABLE | Spliterator.NONNULL | Spliterator.ORDERED) {
        private Status status = Status.BEFORE_ARRAY;

        @Override
        public boolean tryAdvance(final Consumer<? super T> action) {
            for ( ; ; ) {
                switch ( status ) {
                case BEFORE_ARRAY:
                    jsonTokener.next('[');
                    status = Status.IN_ARRAY;
                    continue;
                case IN_ARRAY:
                    switch ( jsonTokener.nextClean() ) {
                    case ',':
                        return true;
                    case ']':
                        status = Status.AFTER_ARRAY;
                        return false;
                    default:
                        jsonTokener.back();
                        @SuppressWarnings("unchecked")
                        final T value = (T) jsonTokener.nextValue();
                        action.accept(value);
                        continue; // or return true?
                    }
                case AFTER_ARRAY:
                    throw new IllegalStateException();
                default:
                    throw new AssertionError(status);
                }
            }
        }
    }, false);
}

private enum Status {

    BEFORE_ARRAY,
    IN_ARRAY,
    AFTER_ARRAY

}

它所做的一切正在改變一些JSON令牌流流的東西（因為org.json物件模型不建議一個共同的基類）。. 如果您已經有JSONArray緩沖，則可以使用以下內容進行流式傳輸：Convert Iterable to Stream using Java 8 JDK

接下來，只需使用上面決議的流中的分組收集器：

final Collector<JSONObject, ?, Map<List<String>, Double>> collector = Collectors.groupingBy(
        // your groups for (col1. col2)
        row -> List.of(row.getString("col1"), row.getString("col2")),
        // your aggregating SUM for col3
        Collectors.summingDouble(row -> row.getDouble("col3"))
);
Assertions.assertEquals(
        Map.of(List.of("c1", "r2"), 8874.0, List.of("c1", "r1"), 13442.0),
        JsonStreams.<JSONObject>asStream(new JSONTokener(reader))
                .collect(collector)
);

就是這樣SUM。AVG結果可以通過使用來完成Collectors.averagingDouble。

標籤：爪哇 json 流

上一篇：從JSON檔案中洗掉特定資料

下一篇：jq不列印select()回傳null的行