詳解Java Stream的分組和聚合
譯文譯者 | 翟珂
審校 | 孫淑娟 梁策
當(dāng)我們將一個集合中的元素分組后，我們可以對分組內(nèi)元素的字段進(jìn)行聚合，執(zhí)行有意義的操作，幫助我們分析數(shù)據(jù)。比如相加，取平均數(shù)，或最大/最小值。此外，還可以用Java Stream和Collectors輕松完成這些字段的聚合。文檔中提供了這些計算的簡單例子。
當(dāng)然，還有更復(fù)雜的聚合，如加權(quán)平均數(shù)、幾何平均數(shù)。另外，可能還需要對幾個字段同時進(jìn)行聚合。在這篇文章中，我們將展示如何使用 Java Stream更快地解決這類問題，這個框架使我們能夠高效地處理大量數(shù)據(jù)。
假設(shè)讀者已對Java Streams和Collectors類有基本的了解：
問題示例
舉一個簡單的例子來展示它的用途，這個例子會盡量通俗，方便概括。一個由TaxEntry實(shí)體構(gòu)成的集合(稅收)，實(shí)體代碼定義如下：
public class TaxEntry {
private String state;
private String city;
private int numEntries;
private double price;
//Constructors, getters, hashCode, equals etc
}
計算每個城市的稅目總數(shù)非常簡單：
Map<String, Integer> totalNumEntriesByCity =
taxes.stream().collect(Collectors.groupingBy(TaxEntry::getCity,
Collectors.summingInt(TaxEntry::getNumEntries)));
Collectors.groupingBy需要兩個參數(shù)：一個分類函數(shù)來做分組條件，一個收集器來做分組后流的組內(nèi)聚合。在這我們使用TaxEntry::getCity作為分類條件。使用Collectors::summingInt方法來處理分組后的流，它返回一個Collector收集器，即對每組的稅目數(shù)進(jìn)行合計。
如果想要進(jìn)行復(fù)合分組，事情就有點(diǎn)復(fù)雜了。例如，在前面的問題中，去求每個省和城市的總稅目數(shù)，我們先定義方法：
record StateCityGroup(String state, String city) {}
注意，我們使用的是一個Java record，這是一種定義不可變數(shù)據(jù)類的簡潔方式。Java編譯器會為我們生成類的getter、setter、hashCode、equals和toString方法。這樣就可以很簡單地解決問題：
Map<StateCityGroup, Integer> totalNumEntriesForStateCity =
taxes.stream().collect(groupingBy(p -> new StateCityGroup(p.getState(), p.getCity()),
Collectors.summingInt(TaxEntrySimple::getNumEntries))
);
我們使用lambda表達(dá)式來設(shè)置分類函數(shù)，創(chuàng)建一個新的StateCityGroup類用來封裝每個省的城市。分組后流的收集器與之前一致。
備注：為了簡潔起見，在代碼示例中，我們假設(shè)Collectors類的所有方法都是靜態(tài)導(dǎo)入的。
如果想同時做幾個聚合，就變得復(fù)雜了。例如，找到一個給定的省和城市的稅目數(shù)和平均價格的總和，框架沒有提供一個簡單的方法。
為了解決這個問題，我們從之前的聚合中得到啟發(fā)，定義一個record，封裝所有需要聚合的字段。
record TaxEntryAggregation (int totalNumEntries, double averagePrice ) {}
現(xiàn)在，我們該怎么同時對這兩個字段進(jìn)行聚合呢?那就是做兩次流收集，分別找到每一個聚合，如下面代碼：
Map<StateCityGroup, TaxEntryAggregation> aggregationByStateCity = taxes.stream().collect(
groupingBy(p -> new StateCityGroup(p.getState(), p.getCity()),
collectingAndThen(Collectors.toList(),
list -> {int entries = list.stream().collect(
summingInt(TaxEntrySimple::getNumEntries));
double priceAverage = list.stream().collect(
averagingDouble(TaxEntrySimple::getPrice));
return new TaxEntryAggregation(entries, priceAverage);})));
分組和以前一樣，但對于分組后流，我們使用Collectors::collectionAndThen進(jìn)行聚合。這個函數(shù)需要兩個參數(shù)：
- 我們將第一次分組的流轉(zhuǎn)換為一個集合(使用Collectors::toList())。
- 我們使用一個lambda表達(dá)式來結(jié)束函數(shù)，從上一步的集合中創(chuàng)建兩個不同的流來做聚合，并存在TaxEntryAggregation類中返回。
如果我們想同時做更多的字段聚合，那么我們將增加后續(xù)流集合中的流數(shù)量。這樣代碼就會變得效率低下，代碼冗余。所以我們應(yīng)該尋找更好的替代方案。
還有一個問題，通常我們在使用Collectors類時，可以做的聚合類型有限。而且求和、求平均和歸納只提供了對integer、long和double類型的支持。如果我們有更復(fù)雜的類型如BigInteger或BigDecimal時，該怎么辦?
更糟的是，歸納方法只提供了min、max、count、sum和average的統(tǒng)計。如果我們想進(jìn)行更復(fù)雜的計算，如加權(quán)平均數(shù)或幾何平均數(shù)，怎么辦?
有些人會說，我們可以編寫自定義的收集器(Collectors)，但這需要深刻理解收集器的接口和對流式收集器流程。不如直接使用Collectors類中的內(nèi)置方法。在下一節(jié)中，我們將解決這些問題。
復(fù)雜的多重聚合：一種解決方法
針對上面的問題，我們寫一個例子。假設(shè)我們有實(shí)體：
public class TaxEntry {
private String state;
private String city;
private BigDecimal rate;
private BigDecimal price;
record StateCityGroup(String state, String city) {
}
//Constructors, getters, hashCode/equals etc
}
我們首先要思考的是，對于每個不同的<省-城市>，我們?nèi)绾文苷业蕉惸康目倲?shù)以及稅率和價格的乘積的總和(∑(稅率*價格))。其中需要注意的點(diǎn)是使用BigDecimal進(jìn)行多字段聚合。
與上一節(jié)一樣，我們定義了一個封裝聚合指標(biāo)的類。
record RatePriceAggregation(int count, BigDecimal ratePrice) {}
對于分組后的簡單聚合，一個高效的方法是Collectors::toMap。
Map<StateCityGroup, RatePriceAggregation> mapAggregation = taxes.stream().collect(
toMap(p -> new StateCityGroup(p.getState(), p.getCity()),
p -> new RatePriceAggregation(1, p.getRate().multiply(p.getPrice())),
(u1,u2) -> new RatePriceAggregation( u1.count() + u2.count(), u1.ratePrice().add(u2.ratePrice()))
));
Collectors::toMap需要三個參數(shù)：
- 第一個參數(shù)是一個lambda表達(dá)式，用于生成Map的key。這個函數(shù)創(chuàng)建StateCityGroup對象作為key。這將按<省+城市>元素進(jìn)行分組。
- 第二個參數(shù)產(chǎn)生Map的value。在示例中，我們創(chuàng)建了一個RatePriceAggregation對象，初始化：1個，稅率與價格的乘積。
- 最后一個參數(shù)是一個二進(jìn)制運(yùn)算器，用于合并相同key(省-城市)的value值。然后將計數(shù)和價格相加進(jìn)行聚合。
下面造一些數(shù)據(jù)來進(jìn)行測試：
List<TaxEntry> taxes = Arrays.asList(
new TaxEntry("New York", "NYC", BigDecimal.valueOf(0.2), BigDecimal.valueOf(20.0)),
new TaxEntry("New York", "NYC", BigDecimal.valueOf(0.4), BigDecimal.valueOf(10.0)),
new TaxEntry("New York", "NYC", BigDecimal.valueOf(0.6), BigDecimal.valueOf(10.0)),
new TaxEntry("Florida", "Orlando", BigDecimal.valueOf(0.3), BigDecimal.valueOf(13.0)));
從上面的map中獲取紐約的結(jié)果：
System.out.println("New York: " + mapAggregation.get(new StateCityGroup("New York", "NYC")));
輸出結(jié)果：
New York: RatePriceAggregation[count=3, ratePrice=14.00]
這是一種解決方法，處理了多個字段和非原始數(shù)據(jù)類型(在我們的例子中為BigDecimal)的分組和聚集。但是，它的缺點(diǎn)是你不能進(jìn)行其他最終結(jié)果的聚合，比如不能做任何形式的平均數(shù)。
如果要計算<稅率-價格>的加權(quán)平均數(shù)，以及每個<省-城市>的所有價格的總和。我們需要先計算屬于每個<省-城市>的所有稅目的稅率和價格的乘積之和，然后除以每種情況的總稅目數(shù)n。1/n ∑(費(fèi)率*價格)。
我們定義一個含有總價的實(shí)體類。
record TaxEntryAggregation(int count, BigDecimal weightedAveragePrice, BigDecimal totalPrice) {}
然后我們解決上述問題：
Map<StateCityGroup, TaxEntryAggregation> groupByAggregation = taxes.stream().collect(
groupingBy(p -> new StateCityGroup(p.getState(), p.getCity()),
mapping(p -> new TaxEntryAggregation(1, p.getRate().multiply(p.getPrice()), p.getPrice()),
collectingAndThen(reducing(new TaxEntryAggregation(0, BigDecimal.ZERO, BigDecimal.ZERO),
(u1,u2) -> new TaxEntryAggregation(u1.count() + u2.count(),
u1.weightedAveragePrice().add(u2.weightedAveragePrice()),
u1.totalPrice().add(u2.totalPrice()))
),
u -> new TaxEntryAggregation(u.count(),
u.weightedAveragePrice().divide(BigDecimal.valueOf(u.count()),
2, RoundingMode.HALF_DOWN),
u.totalPrice())
)
)
));
這段代碼有些復(fù)雜，但有效地解決了問題。下面詳細(xì)講解一下：
- Collectors::groupingBy
1. 我們創(chuàng)建一個StateCityGroup對象用于分組
2. 對于分組后流，我們調(diào)用Collectors::mapping方法
- 第一個參數(shù)，將分組的<省-城市>稅收類轉(zhuǎn)換為TaxEntryAggregation對象，然后初始化：個數(shù)為1，稅率乘以價格，價格。
- 對于后續(xù)流，我們調(diào)用Collectors::collectionAndThen方法進(jìn)行整理轉(zhuǎn)換。
1.調(diào)用Collectors::reducing
- 創(chuàng)建一個有值的TaxEntryAggregation類防止空值。
- Lambda表達(dá)式實(shí)現(xiàn)reducing方法，并返回TaxEntryAggregation對象，并進(jìn)行相應(yīng)字段的聚合。
2.歸納轉(zhuǎn)換，使用前一個reducing中計算的個數(shù)計算平均數(shù)，并返回最終的TaxEntryAggregation。
這個方法不僅可以同時對多個字段進(jìn)行聚合，而且還可以分幾個階段進(jìn)行復(fù)雜的計算。
所以這是一個去解決這類問題的簡單方法。歸納一下就是：定義一個封裝了所有需要聚合的字段的record，使用Collectors::mapping來初始化記錄，然后使用Collectors::collectionAndThen來做二次處理和最終聚合。
與上一節(jié)一樣，我們可以得到紐約的聚合結(jié)果：
System.out.println("Finished aggregation: " + groupByAggregation.get(new StateCityGroup("New York", "NYC")));
結(jié)果：
Finished aggregation: TaxEntryAggregation[count=3, weightedAveragePrice=4.67, totalPrice=40.0]
備注：由于TaxEntryAggregation是一條Java record，且是不可改變的，所以可以使用stream collector庫來并行流計算。
結(jié)論
我們編寫了幾個復(fù)雜的多字段分組聚合示例，其中包括非原始數(shù)據(jù)類型的多字段聚合和跨字段聚合計算。這些表明了可以通過Java Stream和Collectors API及record集合來高效的處理大量數(shù)據(jù)。
譯者介紹
翟珂，51CTO社區(qū)編輯，目前在杭州從事軟件研發(fā)工作，做過電商、征信等方面的系統(tǒng)，享受分享知識的過程，充實(shí)自己的生活。
原文標(biāo)題：Grouping and Aggregations With Java Streams，作者：Manu Barriola
