批量写入操作

在此页面上

Overview

样本数据
定义写入操作
插入操作
更新操作
替换操作
删除操作
执行批量操作
自定义批量写入操作
返回值
更多信息
API 文档

Overview

本指南向您展示如何使用Scala驾驶员执行批量写入操作，从而在单个数据库调用中对数据进行多项更改。

考虑需要为同一任务插入文档、更新文档和删除文档的情况。如果使用单独的写入方法执行每种类型的操作，则每次写入都会单独访问数据库。您可以使用批量写入操作来优化应用程序对服务器的调用次数。

样本数据

本指南中的示例使用restaurants sample_restaurantsAtlas示例数据集的数据库中的集合。要从Scala应用程序访问权限此集合，请创建一个连接到AtlasMongoClient 集群的，并将以下值分配给database 和collection 变量：

val database: MongoDatabase = mongoClient.getDatabase("sample_restaurants")
val collection: MongoCollection[Document] = database.getCollection("restaurants")

要学习；了解如何创建免费的MongoDB Atlas 群集并加载示例数据集，请参阅Atlas入门指南。

定义写入操作

对于您要执行的每个写入操作，创建以下操作类之一的相应实例，这些操作类从通用WriteModel类继承：

InsertOneModel
UpdateOneModel
UpdateManyModel
ReplaceOneModel
DeleteOneModel
DeleteManyModel

然后，将这些实例的列表传递给bulkWrite()方法。

以下部分介绍如何创建和使用上述类的实例。执行批量操作部分演示了如何将模型列表传递给bulkWrite()方法以执行批量操作。

插入操作

要执行插入操作，请创建一个InsertOneModel实例并指定要插入的文档。

以下示例创建了一个InsertOneModel实例：

val insertOneModel = InsertOneModel(
    Document("name" -> "Blue Moon Grill",
             "borough" -> "Brooklyn",
             "cuisine" -> "American")
)

要插入多个文档，请为每个文档创建一个InsertOneModel实例。

重要

执行批量操作时， InsertOneModel无法插入集合中已存在的具有_id的文档。在这种情况下，驾驶员会抛出MongoBulkWriteException 。

更新操作

要更新文档，请创建UpdateOneModel的实例并传递以下参数：

查询过滤，指定用于匹配集合中文档的条件。
要执行的更新操作。有关更新操作的更多信息，请参阅MongoDB Server手册中的字段更新操作指南。

以下示例创建了一个UpdateOneModel实例：

val updateOneFilter = equal("name", "White Horse Tavern")
val updateOneDoc = set("borough", "Queens")
val updateOneModel = UpdateOneModel(updateOneFilter, updateOneDoc)

如果多个文档与 UpdateOneModel实例中指定的查询过滤匹配，则该操作会更新第一个结果。您可以在 UpdateOptions实例中指定排序，以便在驾驶员执行更新操作之前对匹配的文档应用，如以下代码所示：

val options = UpdateOptions.sort(ascending("name"))

要更新多个文档，请创建UpdateManyModel的实例并传递与UpdateOneModel相同的参数。 UpdateManyModel类指定与查询过滤匹配的所有文档的更新。

以下示例创建了一个UpdateManyModel实例：

val updateManyFilter = equal("name", "Wendy's")
val updateManyDoc = set("cuisine", "Fast food")
val updateManyModel = UpdateOneModel(updateManyFilter, updateManyDoc)

替换操作

替换操作会删除指定文档的所有字段和值，并将其替换为您指定的新字段和值。要执行替换操作，请创建 ReplaceOneModel 的实例并传递以下参数：

查询过滤，指定用于匹配集合中文档的条件
指定要插入的新字段和值的替换文档

以下示例创建了一个ReplaceOneModel实例：

val replaceFilter = equal("name", "Cooper Town Diner")
val replaceDoc = Document("name" -> "Smith Town Diner",
                          "borough" -> "Brooklyn",
                          "cuisine" -> "American")
val replaceOneModel = ReplaceOneModel(replaceFilter, replaceDoc)

如果多个文档与 ReplaceOneModel实例中指定的查询过滤匹配，则该操作将替换第一个结果。您可以在 ReplaceOptions实例中指定排序，以便在驾驶员执行替换操作之前对匹配的文档应用顺序，如以下代码所示：

val options = ReplaceOptions.sort(ascending("name"))

提示

替换多个文档

要替换多个文档，请为每个文档创建一个ReplaceOneModel实例。

删除操作

要删除文档，请创建DeleteOneModel的实例并传递查询过滤，指定要删除的文档。 DeleteOneModel实例提供了仅删除与查询过滤匹配的第一个文档的说明。

以下示例创建了一个DeleteOneModel实例：

val deleteOneModel = DeleteOneModel(equal("name", "Morris Park Bake Shop"))

要删除多个文档，请创建DeleteManyModel实例并传递查询过滤，指定要删除的文档。 DeleteManyModel的实例提供了删除与查询过滤匹配的所有文档的说明。

以下示例创建了一个DeleteManyModel实例：

val deleteManyModel = DeleteManyModel(equal("cuisine", "Experimental"))

执行批量操作

为要执行的每个操作定义模型实例后，将这些实例的列表传递给bulkWrite()方法。默认，该方法按照模型列表指定的顺序运行操作。

以下示例使用bulkWrite()方法执行多个写入操作：

val insertOneModel = InsertOneModel(
    Document("name" -> "Red's Pizza",
             "borough" -> "Brooklyn",
             "cuisine" -> "Pizzeria")
)
val updateOneModel = UpdateOneModel(equal("name", "Moonlit Tavern"), set("borough", "Queens"))
val deleteManyModel = DeleteManyModel(equal("name", "Crepe"))
val writes = Seq(insertOneModel, updateOneModel, deleteManyModel)
val observable = collection.bulkWrite(writes)
observable.subscribe(
  (result: BulkWriteResult) => println(s"Success: $result"),
  (error: Throwable) => println(s"Error: ${error.getMessage}"),
  () => println("Completed")
)

Success: AcknowledgedBulkWriteResult{insertedCount=1, matchedCount=1, removedCount=1,
modifiedCount=1, upserts=[], inserts=[BulkWriteInsert{index=0, id=BsonObjectId{value=...}}]}
Completed

如果任何写入操作失败，则Scala驾驶员将引发 BulkWriteError 并且不会执行任何进一步的操作。 BulkWriteError 提供了一个 details 项，其中包括失败的操作以及有关异常的详细信息。

注意

当驾驶员运行批量操作时，它会使用目标集合的写关注（write concern）。无论执行顺序如何，驾驶员在尝试所有操作后都会报告所有写关注（write concern）错误。

自定义批量写入操作

bulkWrite() 方法可以选择接受 BulkWriteOptions 参数，该参数指定可用于配置批量写入操作的选项。如果不指定任何选项，驾驶员将使用默认设置执行批量操作。

下表描述了可用于配置BulkWriteOptions实例的 setter 方法：

方法	说明
`ordered()`	If `true`, the driver performs the write operations in the order provided. If an error occurs, the remaining operations are not attempted. If `false`, the driver performs the operations in an arbitrary order and attempts to perform all operations. Defaults to `true`.
`bypassDocumentValidation()`	Specifies whether the update operation bypasses document validation. This lets you update documents that don't meet the schema validation requirements, if any exist. For more information about schema validation, see Schema Validation in the MongoDB Server manual. Defaults to `false`.
`comment()`	Sets a comment to attach to the operation.
`let()`	Provides a map of parameter names and values to set top-level variables for the operation. Values must be constant or closed expressions that don't reference document fields.

以下代码创建选项并将 ordered 选项设置为 false，以指定无序批量写入。然后，该示例使用 bulkWrite() 方法执行批量操作：

val options = BulkWriteOptions().ordered(false)
val observable = collection.bulkWrite(writes, options)

如果无序批量写入中的任何写入操作失败，则Scala驾驶员仅在尝试所有操作后才会报告错误。

注意

无序批量操作不保证执行顺序。为了优化运行时间，顺序可以与您列出的方式不同。

返回值

bulkWrite() 方法返回一个包含 BulkWriteResult 的 SingleObservable对象。您可以通过订阅可观察对象并使用以下方法从 BulkWriteResult实例访问权限信息：

方法	说明
`wasAcknowledged()`	Indicates if the server acknowledged the write operation.
`getDeletedCount()`	The number of documents deleted, if any.
`getInsertedCount()`	The number of documents inserted, if any.
`getInserts()`	The list of inserted documents, if any.
`getMatchedCount()`	The number of documents matched for an update, if applicable.
`getModifiedCount()`	The number of documents modified, if any.
`getUpserts()`	The list of upserted documents, if any.