MongoDB7.0--SpringBoot聚合操作

发表于 更新于 分类于 ， ， 置顶 精品
阅读次数： 本文字数： 3.4k 阅读时长 ≈ 12 分钟

摘要

聚合操作简介

  • 聚合操作允许用户处理多个文档并返回计算结果

  • 聚合操作包含三类

    • 单文档聚合：针对单个集合，如：db.collection.countDocument(), db.collection.distinct()
    • 聚合管道：它可以作用在一个或几个集合上，对集合中的数据进行的一系列运算，并将这些数据转化为用户期望的形式，本文主要介绍管道操作
    • MapReduce：从MongoDB 5.0开始，map-reduce操作已被弃用，本文不做介绍。

聚合管道

  • 聚合管道是MongoDB中非常强大的功能，它允许用户将多个操作组合在一起，以实现复杂的数据处理。

  • 从效果而言，聚合管道相当于 SQL 查询中的 GROUP BY、 LEFT OUTER JOIN 、 AS等。

  • 整个聚合运算过程称为管道（Pipeline），它是由多个阶段（Stage）组成的， 每个管道：

    • 接受一系列文档（原始数据）
    • 每个阶段对这些文档进行一系列运算
    • 结果文档输出给下一个阶段

  • 聚合管道操作语法

    • pipelines 一组数据聚合阶段。除$out、$Merge和$geonear阶段之外，每个阶段都可以在管道中出现多次。
    • options 可选，聚合操作的其他参数。包含：查询计划、是否使用临时文件、 游标、最大操作时间、读写策略、强制索引等等
    1
    2
    pipeline = [$stage1, $stage2, ...$stageN];
    db.collection.aggregate(pipeline, {options})

    聚合管道

  • 常用的聚合阶段运算符

阶段运算符 描述 SQL等价运算符
$match 过滤文档 WHERE
$project 投影，改变文档的形状和内容 SELECT filedName AS newName
$group 将文档分组 GROUP BY
$sort 对文档进行排序 ORDER BY
$limit 限制结果集的大小 LIMIT
$skip 跳过指定数量的文档 OFFSET
$unwind 展开数组 -
$lookup 从其他集合中获取相关文档,左外连接 LEFT OUTER JOIN
$out 将结果集输出到新的集合 -
$geoNear 按照地理位置附近的顺序返回文档 -
$graphLookup 执行递归查询 -
$addFields 添加新字段 -
$bucket 根据指定条件将文档分组成桶 -
$facet 允许在单个聚合阶段内执行多个独立的子聚合 -

  • 聚合表达式

获取字段信息

1
2
$<field>  ： 用 $ 指示字段路径
$<field>.<subfield>  ： 使用 $  和 .  来指示内嵌文档的路径

常量表达式

1
$literal :<value> ： 指示常量 <value>

系统变量表达式

1
2
$$<variable>  使用 $$ 指示系统变量
$$CURRENT  指示管道中当前操作的文档

聚合管道示例

  • 本示例使用SpringBoot实现MongoDB7.0的聚合操作

  • 初始化数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
/*
var tags = ["nosql","mongodb","document","developer","popular"];
var types = ["technology","sociality","travel","novel","literature"];
var books=[];
for(var i=0;i<50;i++){
    var typeIdx = Math.floor(Math.random()*types.length);
    var tagIdx = Math.floor(Math.random()*tags.length);
    var tagIdx2 = Math.floor(Math.random()*tags.length);
    var favCount = Math.floor(Math.random()*100);
    var username = "xx00"+Math.floor(Math.random()*10);
    var age = 20 + Math.floor(Math.random()*15);
    var book = {
        title: "book-"+i,
        type: types[typeIdx],
        tag: [tags[tagIdx],tags[tagIdx2]],
        favCount: favCount,
        author: {name:username,age:age}
    };
    books.push(book)
}
db.books.insertMany(books);
*/
String[] tags = {"nosql", "mongodb", "document", "developer", "popular"};
String[] types = {"technology", "sociality", "travel", "novel", "literature"};
List<Document> books = new ArrayList<>();

Random random = new Random();
for (int i = 0; i < 50; i++) {
    int typeIdx = random.nextInt(types.length);
    int tagIdx = random.nextInt(tags.length);
    int tagIdx2 = random.nextInt(tags.length);
    int favCount = random.nextInt(100);
    String username = "xx00" + random.nextInt(10);
    int age = 20 + random.nextInt(15);

    Document book = new Document("title", "book-" + i)
            .append("type", types[typeIdx])
            .append("tag", List.of(tags[tagIdx], tags[tagIdx2]))
            .append("favCount", favCount)
            .append("author", new Document("name", username).append("age", age));
    books.add(book);
}

mongoTemplate.insert(books, "books");

$project:投影操作

  • 将原始字段投影成指定名称， 如将集合中的 title 投影成 name

1
2
3
4
5
6
7
8
9
10
// db.books.aggregate([{$project:{name:"$title"}}])
ProjectionOperation projectOperation = Aggregation.project()
        .andExpression("title").as("name");
Aggregation aggregation = Aggregation.newAggregation(projectOperation);
System.out.println(aggregation);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

for (Document document : result) {
    System.out.println(document.toJson());
}

  • 剔除不需要的字段

1
2
3
4
5
6
7
8
9
10
11
12
13
// db.books.aggregate([{$project:{name:"$title",_id:0,type:1,author.name:1}}])
ProjectionOperation projectOperation = Aggregation.project()
        .andExclude("_id")
        .andInclude("type")
        .andExpression("author.name").as("author.name") // 这里必须使用as重新命名，否则只会用name做字段名称，这就和后面的title的别名冲突了
        .andExpression("title").as("name");
Aggregation aggregation = Aggregation.newAggregation(projectOperation);
System.out.println(aggregation);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

for (Document document : results) {
    System.out.println(document.toJson());
}

$match:过滤操作

  • 过滤出指定条件的文档

1
2
3
4
5
6
7
8
// db.books.aggregate([{$match:{type:"technology"}}])
MatchOperation matchOperation = Aggregation.match(Criteria.where("type").is("technology"));
Aggregation aggregation = Aggregation.newAggregation(matchOperation);
System.out.println(aggregation);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();
for (Document document : results) {
    System.out.println(document.toJson());
}

  • 组合其它管道

1
2
3
4
5
6
7
8
9
10
/*
db.books.aggregate([
    {$match:{type:"technology"}},
    {$project:{name:"$title",_id:0,type:1,author:{name:1}}}
])
*/
MatchOperation matchOperation = Aggregation.match(Criteria.where("type").is("technology"));
ProjectionOperation projectOperation = Aggregation.project("title").andExclude("_id").andInclude("type").and("author.name").as("author.name");
Aggregation aggregation = Aggregation.newAggregation(matchOperation, projectOperation);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

$count:计数操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
/*
db.books.aggregate([
    {$match:{type:"technology"}},
    {$count: "type_count"}
])
*/
// MatchOperation to filter documents by type: "technology"
MatchOperation matchAggregation = Aggregation.match(Criteria.where("type").is("technology"));
// CountOperation to count the matched documents
CountOperation countAggregation = Aggregation.count().as("type_count");
// Combine the match and count operations
Aggregation aggregation = Aggregation.newAggregation(matchAggregation, countAggregation);
System.out.println(aggregation);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

$group:分组操作

  • 按指定的表达式对文档进行分组，并将每个不同分组的文档输出到下一个阶段。输出文档包含一个_id字段，该字段按键包含不同的组。

  • 输出文档还可以包含计算字段，该字段保存由$group的_id字段分组的一些accumulator表达式的值。 $group不会输出具体的文档而只是统计信息。

  • 语法

1
2
3
4
{ $group: { _id: <expression>, <field1>: { <accumulator1> : <expression1> }, ... } }
# _id字段是必填的;但是，可以指定_id值为null来为整个输入文档计算累计值。
# 剩余的计算字段是可选的，并使用<accumulator>运算符进行计算。
# _id和<accumulator>表达式可以接受任何有效的表达式。

  • accumulator操作符

名称 描述 类比sql
avg 计算均值 avg
first 返回每组第一个文档，如果有排序，按照排序，如果没有按照默认的存储的顺序的第一个文档。 limit 0,1
last 返回每组最后一个文档，如果有排序，按照排序，如果没有按照默认的存储的顺序的最后个文档。 -
max 根据分组，获取集合中所有文档对应值得最大值。 max
min 根据分组，获取集合中所有文档对应值得最小值。 min
push 将指定的表达式的值添加到一个数组中。 -
addToSet 将表达式的值添加到一个集合中（无重复值，无序）。 -
sum 计算总和 sum
stdDevPop 返回输入值的总体标准偏差（population standard deviation） -
stdDevSamp 返回输入值的样本标准偏差（the sample standard deviation） -

示例

  • book的数量，收藏总数和平均值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
/*
db.books.aggregate([
    {$group:{_id:null,count:{$sum:1},pop:{$sum:"$favCount"},avg:{$avg:"$favCount"}}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.group().count().as("count").sum("favCount").as("pop").avg("favCount").as("avg")
);

AggregationResults<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class);
List<Document> mappedResults = results.getMappedResults();

// Assuming that there will be only one result due to grouping by null (_id: null)
if (!mappedResults.isEmpty()) {
    System.out.println(mappedResults.get(0).toJson());
} else {
    System.out.println("{}");
}

  • 统计每个作者的book收藏总数

1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$group:{_id:"$author.name",pop:{$sum:"$favCount"}}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.group("author.name").sum("favCount").as("pop")
);

AggregationResults<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class);
List<Document> mappedResults = results.getMappedResults();

  • 统计每个作者的每本book的收藏数

1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$group:{_id:{name:"$author.name",title:"$title"},pop:{$sum:"$favCount"}}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.group("author.name", "title").sum("favCount").as("pop")
);

AggregationResults<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class);
List<Document> mappedResults = results.getMappedResults();

  • 每个作者的book的type合集

1
2
3
4
5
6
7
8
9
/*
db.books.aggregate([
    {$group:{_id:"$author.name",types:{$addToSet:"$type"}}}
])
*/
GroupOperation groupOperation = Aggregation.group("author.name").addToSet("type").as("types");
Aggregation aggregation = Aggregation.newAggregation(groupOperation);
AggregationResults<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class);
List<Document> mappedResults = results.getMappedResults();

$unwind

  • 可以将数组拆分为单独的文档

  • 语法

1
2
3
4
5
6
7
8
9
10
11
{
  $unwind:
    {
      #要指定字段路径，在字段名称前加上$符并用引号括起来。
      path: <field path>,
      #可选,一个新字段的名称用于存放元素的数组索引。该名称不能以$开头。
      includeArrayIndex: <string>,
      #可选，default :false，若为true,如果路径为空，缺少或为空数组，则$unwind输出文档
      preserveNullAndEmptyArrays: <boolean>
 }
}

示例

  • 姓名为xx006的作者的book的tag数组拆分为多个文档

1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$match:{"author.name":"xx006"}},
    {$unwind:"$tag"}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("author.name").is("xx006")),
        Aggregation.unwind("tag")
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

  • 使用includeArrayIndex选项来输出数组元素的数组索引

1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$match:{"author.name":"xx006"}},
    {$unwind:{path:"$tag", includeArrayIndex: "arrayIndex"}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("author.name").is("xx006")),
        Aggregation.unwind("tag", "arrayIndex")
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

  • 每个作者的book的tag合集

1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$unwind:"$tag"},
    {$group:{_id:"$author.name",types:{$addToSet:"$tag"}}}
])
*/
UnwindOperation unwindOperation = Aggregation.unwind("tag");
GroupOperation groupOperation = Aggregation.group("author.name").addToSet("tag").as("types");
Aggregation aggregation = Aggregation.newAggregation(unwindOperation, groupOperation);
AggregationResults<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class);
List<Document> mappedResults = results.getMappedResults();

  • 使用preserveNullAndEmptyArrays选项在输出中包含缺少tag字段，null或空数组的文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 初始化数据，加入一些tag为空数组或不存在tag的文档
db.books.insertMany([
    {
        "title" : "book-51",
        "type" : "technology",
        "favCount" : 11,
        "tag":[],
        "author" : {
            "name" : "xx006",
            "age" : 28
        }
    },
    {
        "title" : "book-52",
        "type" : "technology",
        "favCount" : 15,
        "author" : {
            "name" : "xx006",
            "age" : 28
        }
    }
])
1
2
3
4
5
6
7
8
9
10
11
/*
db.books.aggregate([
    {$match:{"author.name":"xx006"}},
    {$unwind:{path:"$tag", preserveNullAndEmptyArrays: true}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("author.name").is("xx006")),
        Aggregation.unwind("tag", true)
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

$limit/$skip/$sort

  • $limit:限制传递到管道中下一阶段的文档数

  • $skip:跳过传递到管道中下一阶段的文档数

  • $sort:对传递到管道中下一阶段的文档进行排序

示例

  • 姓名为xx006的作者的book的tag数组拆分为多个文档，按照收藏数降序排序，跳过2个文档，取5个文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/*
db.books.aggregate([
    {$match:{"author.name":"xx006"}},
    {$unwind:{path:"$tag", preserveNullAndEmptyArrays: true}},
    {$sort:{"favCount":-1}},
    {$limit : 5 },
    {$skip : 2}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("author.name").is("xx006")),
        Aggregation.unwind("tag", true),
        Aggregation.sort(Sort.by(Sort.Direction.DESC, "favCount")),
        Aggregation.skip(2),
        Aggregation.limit(5)
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();
``
* 统计每个分类的book文档数量
```java
/*
db.books.aggregate([
    {$group:{_id:"$type",total:{$sum:1}}},
    {$sort:{total:-1}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.group("type").count().as("total"),
        Aggregation.sort(Sort.by(Sort.Direction.DESC, "total"))
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

  • 标签的热度排行，标签的热度则按其关联book文档的收藏数（favCount）来计算

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/*
db.books.aggregate([
    {$match:{favCount:{$gt:0}}},
    {$unwind:"$tag"},
    {$group:{_id:"$tag",total:{$sum:"$favCount"}}},
    {$sort:{total:-1}}
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.match(Criteria.where("favCount").gt(0)),
        Aggregation.unwind("tag"),
        Aggregation.group("tag").sum("favCount").as("total"),
        Aggregation.sort(Sort.by(Sort.Direction.DESC, "total"))
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

$bucket

  • $bucket:根据指定的条件和边界，将文档分组到不同的桶中

示例

  • 统计book文档收藏数[0,10),[10,60),[60,80),[80,100),[100,+∞）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
/*
db.books.aggregate([{
    $bucket:{
        groupBy:"$favCount",
        boundaries:[0,10,60,80,100],
        default:"other",
        output:{"count":{$sum:1}}
    }
}])

[
  { _id: 0, count: 3 },
  { _id: 10, count: 27 },
  { _id: 60, count: 11 },
  { _id: 80, count: 11 }
]

groupBy: 指定用于分组的字段。在此示例中，根据 favCount 字段的值进行分组。

boundaries: 定义分桶的边界，即每个桶的范围。文档的值将根据这些边界被分配到不同的桶中。边界应按升序排列。在这个命令中，边界 [0, 10, 60, 80, 100] 表示将文档分成六个桶：小于等于0、大于0小于等于10、大于10小于等于60、大于60小于等于80、大于80小于等于100、大于100。

default: 这是一个可选参数，用于指定默认桶的名称。如果文档的值不在指定的边界范围内，则会被放入该桶中。在这个命令中，如果 favCount 的值不在指定的边界范围内，将会被放入名为 "other" 的桶中。比如这里设置边界 [10, 60, 80, 100]，即表示从10开始分组，则0~10的数据会被分组到名为 "other" 的桶中。
[
  { _id: 10, count: 27 },
  { _id: 60, count: 11 },
  { _id: 80, count: 11 },
  { _id: 'other', count: 3 }
]

output: 定义了每个桶的输出内容。这是一个子文档，它指定了在每个桶中计算的聚合操作。在这个示例中，使用 $sum 聚合操作符来计算每个桶中的文档数量，并将结果存储在名为 "count" 的字段中。
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.bucket("$favCount")
                .withBoundaries(0,10,60,80,100)
                .withDefaultBucket("other")
                .andOutput(context -> new Document("$sum", 1))
                .as("count")
);
List<Document> results = mongoTemplate.aggregate(aggregation, "books", Document.class).getMappedResults();

$lookup

  • $lookup:将文档中的一个字段的值与另一个集合中的文档进行匹配，然后将匹配的文档添加到当前文档中

  • 语法

1
2
3
4
5
6
7
8
db.collection.aggregate([{
    $lookup: {
            from: "<collection to join>",
            localField: "<field from the input documents>",
            foreignField: "<field from the documents of the from collection>",
            as: "<output array field>"
        }
})
名称 描述
from 同一个数据库下等待被Join的集合。
localField 源集合中的match值，如果输入的集合中，某文档没有 localField这个Key（Field），在处理的过程中，会默认为此文档含有 localField：null的键值对。
foreignField 待Join的集合的match值，如果待Join的集合中，文档没有foreignField值，在处理的过程中，会默认为此文档含有 foreignField：null的键值对。
as 为输出文档的新增值命名。如果输入的集合中已存在该值，则会覆盖掉。

  • 其语法功能类似于下面的伪SQL语句

1
2
3
4
5
SELECT *, <output array field>
FROM collection
WHERE <output array field> IN (SELECT *
                               FROM <collection to join>
                               WHERE <foreignField> = <collection.localField>);

示例

  • 准备数据

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 顾客信息
db.customer.insertOne({customerCode:1,name:"customer1",phone:"13112345678",address:"test1"})
db.customer.insertOne({customerCode:2,name:"customer2",phone:"13112345679",address:"test2"})

# 订单信息
db.order.insertOne({orderId:1,orderCode:"order001",customerCode:1,price:200})
db.order.insertOne({orderId:2,orderCode:"order002",customerCode:2,price:400})

# 订单明细
db.orderItem.insertOne({itemId:1,productName:"apples",qutity:2,orderId:1})
db.orderItem.insertOne({itemId:2,productName:"oranges",qutity:2,orderId:1})
db.orderItem.insertOne({itemId:3,productName:"mangoes",qutity:2,orderId:1})
db.orderItem.insertOne({itemId:4,productName:"apples",qutity:2,orderId:2})
db.orderItem.insertOne({itemId:5,productName:"oranges",qutity:2,orderId:2})
db.orderItem.insertOne({itemId:6,productName:"mangoes",qutity:2,orderId:2})

  • 查询顾客的订单信息，关联字段为customerCode顾客号码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/*
db.customer.aggregate([
    {$lookup: {
       from: "order",
       localField: "customerCode",
       foreignField: "customerCode",
       as: "customerOrder"
     }
    }
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.lookup("order", "customerCode", "customerCode", "customerOrder")
);
List<Document> results = mongoTemplate.aggregate(aggregation, "customer", Document.class).getMappedResults();

  • 根据订单信息关联顾客信息和订单明细

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/*
db.order.aggregate([
    {$lookup: {
               from: "customer",
               localField: "customerCode",
               foreignField: "customerCode",
               as: "curstomer"
             }

    },
    {$lookup: {
               from: "orderItem",
               localField: "orderId",
               foreignField: "orderId",
               as: "orderItem"
             }
    }
])
*/
Aggregation aggregation = Aggregation.newAggregation(
        Aggregation.lookup("customer", "customerCode", "customerCode", "curstomer"),
        Aggregation.lookup("orderItem", "orderId", "orderId", "orderItem")
);
List<Document> results = mongoTemplate.aggregate(aggregation, "order", Document.class).getMappedResults();