如何使用MongoDB Scala Driver将聚合结果存入集合

我明白你从Java迁移到Scala时遇到的困惑——Scala Driver的异步、函数式设计和Java Driver的回调/同步风格确实有差异。咱们一步步把你的Java逻辑转成Scala的实现：

首先回顾你的Java核心逻辑：聚合后映射文档（把_id里的Document作为chat字段，保留count），然后把结果存入集合。在Scala Driver里，我们要利用它的Observable（基于Reactive Streams）来处理异步流。

准备工作 #

确保你已经导入了Scala Driver的核心包：

import org.mongodb.scala._
import org.mongodb.scala.model.Aggregates._
import org.mongodb.scala.model.InsertManyOptions
import scala.concurrent.Await
import scala.concurrent.duration._

方式一：批量插入（收集所有结果后一次性插入） #

这种方式适合结果集不大的场景，先把聚合结果全部收集到内存，再批量插入目标集合：

// 假设你已经定义了以下变量：
// 聚合管道：aggregatePipeline: Seq[Bson]
// 源集合：sourceCollection: MongoCollection[Document]
// 目标集合：targetCollection: MongoCollection[Document]

// 1. 执行聚合并映射结果，和Java的map逻辑一致
val aggregatedDocs: Observable[Document] = sourceCollection.aggregate(aggregatePipeline)
  .map(doc => Document("chat" -> doc.get("_id", classOf[Document]), "count" -> doc.get("count")))

// 2. 把异步流转为Future，等待所有聚合结果返回
import scala.concurrent.ExecutionContext.Implicits.global
val docsFuture: Future[List[Document]] = aggregatedDocs.toFuture()

// 3. 批量插入到目标集合，ordered=false允许部分插入失败不中断整个流程
val insertResultFuture: Future[InsertManyResult] = docsFuture.flatMap(docs => 
  targetCollection.insertMany(docs, InsertManyOptions().ordered(false))
)

// 同步场景下可以用Await等待结果，异步场景建议用onComplete处理
Await.result(insertResultFuture, 30.seconds)
println("批量插入完成")

方式二：流式插入（边聚合边插入） #

如果结果集很大，不想占用太多内存，可以用流式处理，每产生一个聚合结果就插入一个（或批量插入）：

逐个插入 #

// 聚合映射后，每个文档对应一次插入操作
val insertObservable: Observable[InsertOneResult] = sourceCollection.aggregate(aggregatePipeline)
  .map(doc => Document("chat" -> doc.get("_id", classOf[Document]), "count" -> doc.get("count")))
  .flatMap(doc => targetCollection.insertOne(doc))

// 订阅流来处理成功、失败和完成事件
insertObservable.subscribe(
  result => println(s"单个文档插入成功，ID: ${result.getInsertedId}"),
  error => println(s"插入失败: ${error.getMessage}"),
  () => println("所有聚合结果插入完成")
)

批量流式插入（每N个文档打包插入） #

如果想减少插入请求次数，可以用bufferSize打包批量插入：

import org.mongodb.scala.ObservableImplicits._

sourceCollection.aggregate(aggregatePipeline)
  .map(doc => Document("chat" -> doc.get("_id", classOf[Document]), "count" -> doc.get("count")))
  .bufferSize(100) // 每100个文档打包成一批
  .flatMap(batch => targetCollection.insertMany(batch))
  .subscribe(
    result => println(s"批量插入成功，共插入${result.getInsertedIds.size()}个文档"),
    error => println(s"批量插入失败: ${error.getMessage}"),
    () => println("所有批量插入任务完成")
  )

关键差异说明 #

• Java Driver的.into(new ArrayList<>(), callback)在Scala里被Observable.toFuture()或者直接订阅Observable替代，因为Scala Driver优先采用异步流处理模式。
• Scala的函数式map操作和Java的lambda逻辑一致，但语法更简洁。
• 异步场景下尽量避免用Await.result()阻塞线程，建议用Future.onComplete或者结合Akka等异步框架处理结果。

内容的提问来源于stack exchange，提问作者thiaguten