DynamoDB一括登録とトランザクション｜BatchWriteItemとの違い

DynamoDBで複数の項目をまとめて書く方法は、速くたくさん入れる「BatchWriteItem（boto3では batch_writer）」と、全部成功か全部取り消しかを保証する「TransactWriteItems」の2系統です。 名前は似ていますが、件数の上限も、途中でうまくいかなかったときの動きも違います。

この記事では、最初に比較表で違いを整理し、AWS CLIとboto3の両方で動くコードと実行結果を載せます。大量投入の例（1,000件の登録）と、在庫の引き当てのように「全部そろって初めて成立する」例の両方を扱います。

確認環境：Ubuntu 24.04 / Python 3.11 / boto3 1.43 / AWS CLI 1.46.1。AWS の API をローカルで再現するツール（moto 5.2.3）で実行しました。掲載するコードは、実際の AWS でそのまま使える形（接続先の指定なし）で書いています。上限値・消費キャパシティなど AWS 側の仕様は、AWS 公式ドキュメントの記載にもとづきます。

3つの書き込み方法の比較表

準備：テーブルと初期データ

在庫テーブル Stock（キー：item_id）と注文テーブル Orders（キー：order_id）を作ります。

aws dynamodb create-table --table-name Stock \
  --attribute-definitions AttributeName=item_id,AttributeType=S \
  --key-schema AttributeName=item_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST --query 'TableDescription.TableStatus'

aws dynamodb create-table --table-name Orders \
  --attribute-definitions AttributeName=order_id,AttributeType=S \
  --key-schema AttributeName=order_id,KeyType=HASH \
  --billing-mode PAY_PER_REQUEST --query 'TableDescription.TableStatus'

"ACTIVE"
"ACTIVE"

BatchWriteItem：AWS CLIでまとめて登録・削除する

--request-items に「テーブル名 → リクエストの配列」のJSONを渡します。stock.json：

{
  "Stock": [
    {"PutRequest": {"Item": {"item_id": {"S": "pen"},      "name": {"S": "ボールペン"}, "qty": {"N": "10"}}}},
    {"PutRequest": {"Item": {"item_id": {"S": "notebook"}, "name": {"S": "ノート"},     "qty": {"N": "3"}}}},
    {"PutRequest": {"Item": {"item_id": {"S": "eraser"},   "name": {"S": "消しゴム"},   "qty": {"N": "0"}}}}
  ]
}

PutとDelete、複数テーブルを1回に混ぜることもできます。mix.json：

{
  "Stock": [
    {"DeleteRequest": {"Key": {"item_id": {"S": "eraser"}}}},
    {"PutRequest": {"Item": {"item_id": {"S": "clip"}, "name": {"S": "クリップ"}, "qty": {"N": "50"}}}}
  ],
  "Orders": [
    {"PutRequest": {"Item": {"order_id": {"S": "o-0001"}, "status": {"S": "new"}}}}
  ]
}

aws dynamodb batch-write-item --request-items file://stock.json --query UnprocessedItems
aws dynamodb batch-write-item --request-items file://mix.json --query UnprocessedItems
aws dynamodb scan --table-name Stock --query 'Items[].[item_id.S,qty.N]' --output text

--query UnprocessedItems を付けると、書き残しがあるときだけ内容が表示されます。今回は空だったので何も表示されず、scan の結果は次のとおりでした。

clip    50
notebook    3
pen 10

BatchWriteItem の決まりごと

AWSのAPIリファレンスに書かれている要点です。

• 全体としては原子的ではありません。 1件ずつのPut・Deleteは原子的ですが、25件のうち20件だけ書けて5件が残る、ということが起こります。残った分は UnprocessedItems に入って返ります。
• Updateはできません。 すでにある項目にPutすると、項目全体が上書きされます。一部の属性だけ変えたいときは UpdateItem を使います。
• 条件式は付けられません。
• 次のどれかに当たると、リクエスト全体が拒否されます（1件も書かれません）。

UnprocessedItems の再送ループ（boto3）

batch_write_item を直接使うときは、25件ずつに分ける処理と、UnprocessedItems がなくなるまで送り直す処理を自分で書きます。AWSは、すぐに送り直すと同じ理由でまた残りやすいため、待ち時間を倍々に延ばす方法（指数バックオフ）を強く勧めています。

# retry.py
import time
import random

def batch_write_all(client, table_name, requests, max_retries=8):
    """requests（PutRequest / DeleteRequest の一覧）を25件ずつ送り、残りがなくなるまで再送する"""
    for start in range(0, len(requests), 25):
        pending = {table_name: requests[start:start + 25]}
        for attempt in range(max_retries + 1):
            res = client.batch_write_item(RequestItems=pending)
            pending = res.get("UnprocessedItems", {})
            if not pending:
                break
            left = sum(len(v) for v in pending.values())
            wait = min(2 ** attempt * 0.1, 5) * random.uniform(0.5, 1.0)
            print(f"  残り {left} 件 → {wait:.2f} 秒待って再送（{attempt + 1}回目）")
            time.sleep(wait)
        else:
            raise RuntimeError(f"再送の上限に達しました: {pending}")

UnprocessedItems は「そのまま次の RequestItems に渡せる形」で返るのがポイントです。

書き残しが返る状況は、botocore に付属する Stubber（決めた応答を返させる仕組み）で作って確かめます。1回目に3件、2回目に1件が残る想定です。

# retry_demo.py
import boto3
from botocore.stub import Stubber
from retry import batch_write_all

# UnprocessedItems が返る状況を、botocore の Stubber で作って再送ループを確かめる
client = boto3.client("dynamodb", region_name="ap-northeast-1",
                      aws_access_key_id="test", aws_secret_access_key="test")
reqs = [{"PutRequest": {"Item": {"article_id": {"S": f"r{i}"}}}} for i in range(5)]

stub = Stubber(client)
stub.add_response("batch_write_item", {"UnprocessedItems": {"Articles": reqs[2:]}},
                  {"RequestItems": {"Articles": reqs}})          # 1回目：3件残る
stub.add_response("batch_write_item", {"UnprocessedItems": {"Articles": reqs[4:]}},
                  {"RequestItems": {"Articles": reqs[2:]}})      # 2回目：1件残る
stub.add_response("batch_write_item", {"UnprocessedItems": {}},
                  {"RequestItems": {"Articles": reqs[4:]}})      # 3回目：すべて完了
with stub:
    batch_write_all(client, "Articles", reqs)
    stub.assert_no_pending_responses()
print("すべて書き込み完了")

  残り 3 件 → 0.09 秒待って再送（1回目）
  残り 1 件 → 0.10 秒待って再送（2回目）
すべて書き込み完了

残った分だけが次のリクエストに渡り、3回目で空になってループを抜けています（待ち時間は乱数を掛けているため実行ごとに変わります）。同じ関数で60件を登録すると、25件・25件・10件の3回に分かれて送られ、テーブルには60件が入りました。

# retry_real.py
import boto3
from retry import batch_write_all

c = boto3.client("dynamodb")
reqs = [{"PutRequest": {"Item": {"article_id": {"S": f"b{i:03}"}, "views": {"N": "0"}}}}
        for i in range(60)]
batch_write_all(c, "Articles", reqs)
print(c.scan(TableName="Articles", Select="COUNT",
             FilterExpression="begins_with(article_id, :p)",
             ExpressionAttributeValues={":p": {"S": "b"}})["Count"])

60

batch_writer：大量投入はこれが手軽（boto3）

Table.batch_writer() は、上の「25件ずつに分ける」「残りを再送する」を内部で行います。with の中で put_item / delete_item を呼ぶだけです。1,000件を登録し、送信されたリクエストの回数も数えてみます。

# bulk.py
import boto3

dynamodb = boto3.resource("dynamodb")
table = dynamodb.create_table(
    TableName="Articles",
    AttributeDefinitions=[{"AttributeName": "article_id", "AttributeType": "S"}],
    KeySchema=[{"AttributeName": "article_id", "KeyType": "HASH"}],
    BillingMode="PAY_PER_REQUEST",
)
table.wait_until_exists()

# 送信したリクエストの回数と件数を数える（確認用）
sent = []
def count(params, **kwargs):
    sent.append(sum(len(v) for v in params["RequestItems"].values()))
dynamodb.meta.client.meta.events.register("provide-client-params.dynamodb.BatchWriteItem", count)

with table.batch_writer() as batch:
    for i in range(1000):
        batch.put_item(Item={"article_id": f"a{i:04}", "title": f"記事{i}", "views": i * 3})

print("リクエスト回数:", len(sent))
print("1回あたりの件数:", sent[0], "…", sent[-1])
print("テーブルの件数:", table.scan(Select="COUNT")["Count"])

リクエスト回数: 40
1回あたりの件数: 25 … 25
テーブルの件数: 1000

1,000件が25件×40回に自動で分かれて送られました。 resource 側のAPIなので、値は {"S": "..."} の型記述子なしでPythonの値をそのまま渡せます。with を抜けるときに、バッファに残った分もまとめて送られます。

同じキーが重なるときは overwrite_by_pkeys

元データに同じキーの行が2つあると、同じ25件の中に入った時点で BatchWriteItem の「同じキーは1回まで」という決まりに当たります。boto3の公式ガイドには、そのときのエラーが次の形で載っています。

botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the BatchWriteItem operation: Provided list of item keys contains duplicates

overwrite_by_pkeys にキー属性の名前を渡すと、バッファの中で同じキーのリクエストを後から来たものに置き換えてから送ります。 実際に何が送信されるかを、指定なしと指定ありで比べます。

# dupkeys.py
import boto3

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("Articles")

def run(**options):
    sent = []
    def record(params, **kwargs):
        sent.append([r["PutRequest"]["Item"] for r in params["RequestItems"]["Articles"]])
    ev = dynamodb.meta.client.meta.events
    ev.register("provide-client-params.dynamodb.BatchWriteItem", record)
    with table.batch_writer(**options) as batch:
        batch.put_item(Item={"article_id": "dup-1", "title": "1回目"})
        batch.put_item(Item={"article_id": "dup-2", "title": "別のキー"})
        batch.put_item(Item={"article_id": "dup-1", "title": "2回目"})
    ev.unregister("provide-client-params.dynamodb.BatchWriteItem", record)
    for req in sent:
        print("  送信:", [(i["article_id"], i["title"]) for i in req])

print("指定なし")
run()
print("overwrite_by_pkeys=['article_id']")
run(overwrite_by_pkeys=["article_id"])

指定なし
  送信: [('dup-1', '1回目'), ('dup-2', '別のキー'), ('dup-1', '2回目')]
overwrite_by_pkeys=['article_id']
  送信: [('dup-2', '別のキー'), ('dup-1', '2回目')]

指定なしでは dup-1 が2件入ったまま送られ、指定ありでは「2回目」だけが残っています。ソートキーのあるテーブルでは overwrite_by_pkeys=["パーティションキー名", "ソートキー名"] と両方を渡します。

TransactWriteItems：全部成功か全部取り消し

在庫の引き当てを1つのトランザクションにする

「注文を登録する」と「在庫を減らす」は、片方だけ成立すると数が合わなくなります。TransactWriteItems にまとめると、どれか1つでも条件を満たさなければ、全部が取り消されます。

• 注文：同じ order_id がまだ存在しないこと（attribute_not_exists）
• 在庫：残数が引き当て数以上であること（qty >= :n）

# tx.py
import boto3
from botocore.exceptions import ClientError

c = boto3.client("dynamodb")

def reserve(order_id, lines):
    """lines = [(item_id, 数量), ...] をまとめて引き当て、注文を1件登録する"""
    items = [{
        "Put": {
            "TableName": "Orders",
            "Item": {"order_id": {"S": order_id}, "status": {"S": "reserved"}},
            "ConditionExpression": "attribute_not_exists(order_id)",
        }
    }]
    for item_id, n in lines:
        items.append({
            "Update": {
                "TableName": "Stock",
                "Key": {"item_id": {"S": item_id}},
                "UpdateExpression": "SET qty = qty - :n",
                "ConditionExpression": "qty >= :n",
                "ExpressionAttributeValues": {":n": {"N": str(n)}},
                "ReturnValuesOnConditionCheckFailure": "ALL_OLD",
            }
        })
    try:
        c.transact_write_items(TransactItems=items)
        print(order_id, "→ 引き当て完了")
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] != "TransactionCanceledException":
            raise
        print(order_id, "→ 取り消し")
        print("  Message:", e.response["Error"]["Message"])
        for i, reason in enumerate(e.response["CancellationReasons"]):
            print(f"  [{i}] {reason}")
        return False

def show():
    rows = c.scan(TableName="Stock")["Items"]
    print("  在庫:", {r["item_id"]["S"]: int(r["qty"]["N"])
                     for r in sorted(rows, key=lambda r: r["item_id"]["S"])})
    print("  注文:", sorted(r["order_id"]["S"] for r in c.scan(TableName="Orders")["Items"]))

show()
reserve("o-1001", [("pen", 4), ("notebook", 2)])
show()
reserve("o-1002", [("pen", 4), ("notebook", 2)])   # notebook が足りない
show()
reserve("o-1001", [("pen", 1)])                    # 同じ注文IDの二重登録
show()

  在庫: {'clip': 50, 'notebook': 3, 'pen': 10}
  注文: ['o-0001']
o-1001 → 引き当て完了
  在庫: {'clip': 50, 'notebook': 1, 'pen': 6}
  注文: ['o-0001', 'o-1001']
o-1002 → 取り消し
  Message: Transaction cancelled, please refer cancellation reasons for specific reasons [None, None, ConditionalCheckFailed]
  [0] {'Code': 'None'}
  [1] {'Code': 'None'}
  [2] {'Item': {'item_id': {'S': 'notebook'}, 'name': {'S': 'ノート'}, 'qty': {'N': '1'}}, 'Code': 'ConditionalCheckFailed', 'Message': 'The conditional request failed'}
  在庫: {'clip': 50, 'notebook': 1, 'pen': 6}
  注文: ['o-0001', 'o-1001']
o-1001 → 取り消し
  Message: Transaction cancelled, please refer cancellation reasons for specific reasons [ConditionalCheckFailed, None]
  [0] {'Code': 'ConditionalCheckFailed', 'Message': 'The conditional request failed'}
  [1] {'Code': 'None'}
  在庫: {'clip': 50, 'notebook': 1, 'pen': 6}
  注文: ['o-0001', 'o-1001']

o-1002 では pen は4本引けるはずでしたが、notebook が足りなかったため、pen の在庫も注文の登録も行われていません（在庫は pen: 6 のまま、注文に o-1002 はなし）。これが BatchWriteItem との一番の違いです。

CancellationReasons の読み方

TransactionCanceledException の応答には CancellationReasons が入っています。読み方は3点です。

上の出力では、[2] の Item に qty: 1 が見えるので、「notebook は残り1で、2は引けなかった」と利用者に伝えられます。もう一度読みに行かなくて済むのが利点です。

Code に入る主な値です（AWSのAPIリファレンスより）。

原因が ConditionalCheckFailed だけなら再送せずに利用者へ返し、TransactionConflict やスロットリングなら待って再送する、という分け方ができます。

def should_retry(e):
    codes = {r["Code"] for r in e.response.get("CancellationReasons", [])}
    return bool(codes & {"TransactionConflict", "ProvisionedThroughputExceeded", "ThrottlingError"})

条件不一致で取り消された例外を渡すと False、TransactionConflict を含む例外を渡すと True が返ることを確かめています。

AWS CLI でトランザクションを実行する

tx.json（注文 o-3001 を登録し、clip を30減らす）：

[
  {
    "Put": {
      "TableName": "Orders",
      "Item": {"order_id": {"S": "o-3001"}, "status": {"S": "reserved"}},
      "ConditionExpression": "attribute_not_exists(order_id)"
    }
  },
  {
    "Update": {
      "TableName": "Stock",
      "Key": {"item_id": {"S": "clip"}},
      "UpdateExpression": "SET qty = qty - :n",
      "ConditionExpression": "qty >= :n",
      "ExpressionAttributeValues": {":n": {"N": "30"}}
    }
  }
]

aws dynamodb transact-write-items --transact-items file://tx.json \
  --client-request-token order-o-3001
aws dynamodb get-item --table-name Stock --key '{"item_id":{"S":"clip"}}' --query 'Item.qty.N'

"20"

clip が50から20になりました。同じファイルをトークンなしでもう一度実行すると、注文はすでにあり、clip も30に足りないので、両方の条件が合わずに取り消されます。

aws dynamodb transact-write-items --transact-items file://tx.json

An error occurred (TransactionCanceledException) when calling the TransactWriteItems operation: Transaction cancelled, please refer cancellation reasons for specific reasons [ConditionalCheckFailed, ConditionalCheckFailed]

CLIでは、メッセージ末尾の [...] が CancellationReasons の Code を順番に並べたものです。clip は20のまま変わっていません。

--client-request-token（boto3では ClientRequestToken）は、同じリクエストを重ねて送っても1回分として扱ってもらうための目印です。AWS公式ドキュメントによると、トークンは最初のリクエストから10分間有効で、同じトークンで中身の違うリクエストを送ると IdempotentParameterMismatch が返ります。通信が途切れて結果がわからないまま再送する場面で役立ちます。

トランザクションの決まりごと

件数と重複の決まりを実行して確かめます。

# tx_limits.py
import boto3
from botocore.exceptions import ClientError

c = boto3.client("dynamodb")

def try_tx(label, items):
    try:
        c.transact_write_items(TransactItems=items)
        print(label, "→ OK")
    except ClientError as e:
        print(label, "→", e.response["Error"]["Code"])

def add(n):
    return {"Update": {"TableName": "Stock", "Key": {"item_id": {"S": "pen"}},
                       "UpdateExpression": "SET qty = qty + :n",
                       "ExpressionAttributeValues": {":n": {"N": str(n)}}}}

puts = [{"Put": {"TableName": "Stock", "Item": {"item_id": {"S": f"t{i:03}"}, "qty": {"N": "1"}}}}
        for i in range(101)]

try_tx("同じ項目を2回", [add(1), add(2)])
try_tx("101件", puts)
try_tx("100件", puts[:100])

同じ項目を2回 → ValidationException
101件 → ValidationException
100件 → OK

同じ項目を2回含めたときのメッセージは Transaction request cannot include multiple operations on one item でした。同じ項目に「条件の確認」と「更新」をしたいときは、ConditionCheck を別に置くのではなく、Update 自体の ConditionExpression に条件を書きます（この記事の在庫の例がその形です）。

使い分け：大量投入とトランザクション

AWS公式ドキュメントのベストプラクティスも、必要のない操作をトランザクションにまとめないこと、大量の取り込みには BatchWriteItem を使うことを勧めています。迷ったら「片方だけ成立したら困るか」で決めます。困らないなら batch_writer、困るなら TransactWriteItems です。

100件を超える変更を1つの原子的なまとまりにすることはできません。その場合は、1回のトランザクションに入る単位（注文1件と、その明細が触る在庫）にまとまりを切り直します。

よくある質問

Q. batch_writer の中で1件ずつエラーを拾えますか？ A. batch_writer は25件たまった時点や with を抜ける時点でまとめて送るため、put_item を呼んだ行では結果はまだ出ていません。例外は送信のタイミングで上がります。1件ごとの成否や条件が必要なら、PutItem を1件ずつ呼ぶか、TransactWriteItems を使います。

Q. BatchWriteItem で「なければ登録」はできますか？ A. 条件式が使えないため、できません。すでにある項目は上書きされます。上書きを避けたいときは、attribute_not_exists を付けた PutItem か TransactWriteItems を使います。

Q. トランザクションの読み取り版はありますか？ A. TransactGetItems があります。複数の項目を、同じ時点の整合した状態でまとめて読みたいときに使います。

Q. トランザクションと BatchWriteItem が同じ項目を同時に触るとどうなりますか？ A. AWS公式ドキュメントによると、トランザクションと BatchWriteItem の中の個々の書き込みとの間は直列化されますが、BatchWriteItem 全体を1つの単位としては直列化されません。まとまりとしての整合が必要な側は、トランザクションで書きます。

まとめ

• BatchWriteItem は「速く多く」、TransactWriteItems は「全部成功か全部取り消し」。 目的が違う別の道具。
• 件数上限は BatchWriteItem が25件、TransactWriteItems が100件。 batch_writer は25件ずつ自動で分けて送る。
• BatchWriteItem は UnprocessedItems を見て、待ち時間を延ばしながら再送する。 boto3なら batch_writer に任せられる。
• 同じキーが重なるデータは overwrite_by_pkeys で後勝ちにまとめる。
• TransactionCanceledException は CancellationReasons を順番で読む。 'None' 以外の要素が原因で、ALL_OLD を指定すればその時点の項目も受け取れる。
• トランザクションは書き込み2回分のキャパシティを使う。 片方だけ成立すると困る処理に絞って使う。

参考・出典

• Amazon DynamoDB API Reference「BatchWriteItem」「TransactWriteItems」
• Amazon DynamoDB 開発者ガイド「Amazon DynamoDB Transactions: How it works」
• Boto3 ドキュメント「Amazon DynamoDB」ガイド（Batch writing）

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。