MongoDB(35) - 一括処理（Bulk）④一括登録／一括更新処理(upsert)

今回は、一括登録／一括更新処理(upsert)を行います。

一括登録／一括更新処理(upsert)

順次処理で一括登録／一括更新処理(upsert)を行います。

upsert処理では該当するデータが存在する場合は更新を行い、該当するデータがない場合は登録を行います。

（並列処理で実行したい場合は、initializeOrderedBulkOp()の代わりにinitializeUnorderedBulkOp()を使います。）

処理の詳細は以下の通りです。

  1. Bulk実行タイプを設定。（1行目）
  2. upsertクエリーを設定。（3～6行目）
    前半の3件が更新処理で、最後の1件が登録処理になります。
  3. 一括実行(Bulk実行)。（8行目）

[Mongoシェル]

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
> var bulk = db.member.initializeOrderedBulkOp()

> bulk.find({LastName:'武元'}).upsert().updateOne({$set:{age:12}})  // 更新
> bulk.find({LastName:'松田'}).upsert().updateOne({$set:{age:23}})  // 更新
> bulk.find({LastName:'加藤'}).upsert().updateOne({$set:{age:33}})  // 更新
> bulk.find({LastName:'伊藤'}).upsert().updateOne({$set:{age:43}})  // 追加

> bulk.execute()
BulkWriteResult({
        "writeErrors" : [ ],
        "writeConcernErrors" : [ ],
        "nInserted" : 0,
        "nUpserted" : 1,
        "nMatched" : 3,
        "nModified" : 3,
        "nRemoved" : 0,
        "upserted" : [
                {
                        "index" : 3,
                        "_id" : ObjectId("613e8ff0d1340480894d324d")
                }
        ]
})

> db.member.find()
{ "_id" : ObjectId("613e8e9a0fb0f3e6cf9e2b87"), "LastName" : "武元", "FirstName" : "ゆい", "age" : 12 }
{ "_id" : ObjectId("613e8e9a0fb0f3e6cf9e2b88"), "LastName" : "松田", "FirstName" : "りな", "age" : 23 }
{ "_id" : ObjectId("613e8e9a0fb0f3e6cf9e2b89"), "LastName" : "加藤", "FirstName" : "しほ", "age" : 33 }
{ "_id" : ObjectId("613e8ff0d1340480894d324d"), "LastName" : "伊藤", "age" : 43 }

前半の3ドキュメントに関してはデータが存在していたので、既存データにageフィールドが追加されています。

最後の1ドキュメントに関してはデータが存在していなかったため、新規ドキュメントとしてデータが追加されています。

Pythonで操作

上記の処理をPythonで行うと、次のようになります。

[ソースコード]

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
import pymongo
from pymongo import MongoClient
from pymongo import UpdateOne
from pymongo.errors import BulkWriteError

# MongoDB操作用のインスタンスを作成
client = MongoClient()  # [IPとポートを指定する場合] MongoClient('10.200.243.203', 27017')

# データベースの取得
db1 = client.db1

# 一括入れ替え
requests = [
    UpdateOne({'LastName': '武元'},{'$set':{'age':12}}, upsert=True),
    UpdateOne({'LastName': '松田'},{'$set':{'age':23}}, upsert=True),
    UpdateOne({'LastName': '加藤'},{'$set':{'age':33}}, upsert=True),
    UpdateOne({'LastName': '伊藤'},{'$set':{'age':43}}, upsert=True)
]

try:
    db1.member.bulk_write(requests)                   # 順次処理
    # db.member.bulk_write(requests, ordered=False)   # 並列処理
except BulkWriteError as bwe:
    pprint(bwe.details)

# 入れ替え内容の確認
docs = db1.member.find()
for doc in docs:
    print(doc)

Pythonでupsert処理を行う場合は、UpdateOneを使ってupsert=Trueオプションを追加します。


次回は、一括削除処理(remove)を行います。