DynamoDB Streamsを本番で扱う：順序・重複・失敗処理の設計
DynamoDB StreamsからLambdaへ変更を渡すときは、レコードの形式を知るだけでは足りません。同じ変更をLambdaが再処理する場合や、処理できないレコードがシャード内の後続レコードを止める場合まで考える必要があります。ここでは、注文ステータスの変更を別システムへ反映する例で、ビュータイプから失敗時の復旧まで設計します。
NEW_AND_OLD_IMAGESで変更レコードを読む
注文テーブルの項目 order-001 が PENDING から CONFIRMED に変わったとします。テーブルでDynamoDB Streamsを有効にし、StreamViewType を NEW_AND_OLD_IMAGES にすると、更新前後の項目全体を記録できます。次は必要なフィールドだけを抜き出した概念例です。識別子とシーケンス番号は説明用の値です。
{
"eventID": "example-event-2",
"eventName": "MODIFY",
"dynamodb": {
"Keys": { "orderId": { "S": "order-001" } },
"OldImage": {
"orderId": { "S": "order-001" },
"status": { "S": "PENDING" }
},
"NewImage": {
"orderId": { "S": "order-001" },
"status": { "S": "CONFIRMED" }
},
"SequenceNumber": "200",
"StreamViewType": "NEW_AND_OLD_IMAGES"
}
}
OldImage.status と NewImage.status を比較すれば、この遷移だけを処理できます。INSERT には変更前の項目がなく、REMOVE には変更後の項目がありません。項目を更新するAPIを呼んでもデータが変わらなければ、ストリームレコードは作られません。DynamoDB Streamsの仕様、レコードのイベント種別
StreamViewType |
項目について記録される情報 | この例での判断 |
|---|---|---|
KEYS_ONLY |
キー属性のみ | 対象の注文を特定できるが、状態の遷移は分からない |
NEW_IMAGE |
変更後の項目全体 | 現在の状態を反映する用途に向く |
OLD_IMAGE |
変更前の項目全体 | 変更前の状態を参照する用途に向く |
NEW_AND_OLD_IMAGES |
変更前後の項目全体 |
PENDING → CONFIRMED の遷移を判定できる |
ビュータイプは既存ストリーム上で変更できません。変える場合はストリームを無効にして新しいストリームを作るため、必要な項目情報を最初に決めます。StreamSpecification、DynamoDB Streamsの設定
シャードと同一項目の順序保証
ストリームレコードはシャードに分けられ、各レコードにはシーケンス番号が付きます。DynamoDB Streamsが保証するのは、同じ主キーで識別される項目への変更順です。テーブル全体や、同じパーティションキーを持つ別項目を横断した順序は保証されません。例えば同じ注文への PENDING → CONFIRMED → SHIPPED は変更順に現れますが、別注文との前後関係を一つの列として扱う設計はできません。また、複数項目を一つのDynamoDBトランザクションで更新しても、そのストリームレコードがまとまって届くとは限りません。DynamoDB Streamsの順序保証、トランザクションとストリーム
Lambdaのイベントソースマッピングはシャードからレコードを読みます。ParallelizationFactor を増やして同一シャードの複数バッチを並列処理しても、同一項目の変更順序は維持されます。ただし、次節の再試行により同じレコードを複数回処理する可能性は残ります。LambdaによるDynamoDB Streamsの処理
Lambdaの再試行と重複処理
「各レコードがストリームに一度だけ現れる」というDynamoDB Streamsの保証と、「Lambdaのコンシューマーが各レコードを一度だけ処理する」という保証は別です。Lambdaのイベントソースマッピングは少なくとも1回の処理であり、失敗後の再試行などで同じレコードが再び関数へ渡されます。DynamoDB StreamsとLambdaのベストプラクティス
例えば、CONFIRMED を受けて外部の集計値を1増やした直後に関数が失敗すると、再試行で同じ加算を繰り返すおそれがあります。この処理には冪等性が必要です。項目の現在状態を上書きする処理でも、古い変更で新しい状態を戻さないように、対象側のバージョンや処理済みイベントを照合します。
反映先もDynamoDBであれば、eventID をキーにした処理済み記録を条件付きで作成し、投影先の更新と一つの TransactWriteItems にまとめる方法があります。再配送で処理済み記録の条件が満たされなければ、その条件が原因で失敗したことを確認して重複として扱います。スロットリングや別の条件エラーまで重複扱いにしてはいけません。メール送信などDynamoDB外の副作用は同じトランザクションに含められないため、送信先の冪等キーや別の連携方式が必要です。処理済み記録を期限付きで削除するなら、S3からの後日再処理も含めて重複排除が必要な期間を先に決めます。DynamoDBトランザクション
冪等性・部分バッチ失敗・失敗時の退避先
バッチに複数のレコードが入る場合、関数が例外を返すと、既定ではバッチ全体が失敗として再試行されます。処理できないレコードが一つ残ると、同じシャードの後続処理がレコードの失効まで止まり得ます。DynamoDB Streamsの保持期間は最大24時間です。Lambdaの失敗処理、ストリームの保持期間
ReportBatchItemFailures を有効にすれば、関数は最初に失敗したレコードの dynamodb.SequenceNumber を返せます。Lambdaはその位置から再試行します。複数の失敗番号を返した場合も、最小の番号がチェックポイントになります。先頭から順に処理し、失敗した時点で止めてその番号を返すと、後続を先に処理して再処理する事態を避けられます。部分バッチ失敗の仕様
例えばバッチにシーケンス番号 201、202、203 の順に入っていて、202 で失敗したなら、202 を返して 203 はまだ処理しません。201 は成功としてチェックポイントを進められ、再試行は 202 から始まります。203 まで処理してから 202 の失敗を返すと、203 の副作用も再試行され得ます。
再試行回数やレコードの最大経過時間に上限を設ける場合は、処理しきれなかったイベントの退避先も用意します。SQSやSNSへの失敗通知には呼び出しのメタデータが入り、S3への退避には元の呼び出しレコードも含まれます。後日、元データを調べたり再投入したりする要件があるなら、S3を候補にします。ただし、退避は正常処理ではないため、原因の修正と再処理の手順を別に設計します。DynamoDBイベントソースの失敗先
設定例と障害シナリオ別の確認手順
次は注文ステータスの反映処理を始める際の設計例です。数値はワークロードを測定して決める初期値であり、すべての環境に適した推奨値ではありません。
| 設定 | 例 | 判断理由 |
|---|---|---|
StartingPosition |
TRIM_HORIZON |
マッピング作成時に LATEST を選ぶと、ポーリング開始までの変更を取りこぼす可能性がある |
ParallelizationFactor |
1 |
まず単純な順次処理で始め、滞留を見て並列度を上げる |
FunctionResponseTypes |
ReportBatchItemFailures |
最初の失敗位置から再試行する |
BisectBatchOnFunctionError |
true |
予期しない関数エラーで失敗バッチを分割する |
MaximumRetryAttempts |
3 |
処理できないレコードによるシャードの長時間停止を抑える |
DestinationConfig.OnFailure |
S3バケット | 失敗した呼び出しの元レコードを残す |
MaximumRetryAttempts を小さくするほど後続処理は進みやすくなりますが、修復前にレコードを退避へ送る可能性も高まります。業務上許容できる遅延と再処理手順で値を決めてください。TRIM_HORIZON で開始すると保持期間内の既存レコードも処理対象になるため、対象側の冪等性を先に用意します。S3退避先を使う場合は、Lambdaの実行ロールに必要な書き込み権限も設定します。イベントソースマッピングと開始位置、失敗処理オプション
構築後の確認では、次の順で想定どおりの結果になるかを見ます。
- 同じ
MODIFYレコードを再処理し、対象の状態や集計値が二重に変わらないことを確かめる。 - 処理できないレコードを含むバッチを渡し、失敗したシーケンス番号から再試行されることと、後続の処理がどう進むかを確かめる。
- 再試行の上限を超えた際に失敗先へ記録が届き、退避内容から原因を特定できることを確かめる。
- 処理を意図的に遅らせ、Lambdaの
IteratorAgeが増えることを監視する。レコードは最大24時間で失効するため、滞留を放置しない。Lambdaメトリクス
設計上の判断とまとめ
更新前後の差分で分岐するなら NEW_AND_OLD_IMAGES を選びます。順序は同一項目について保証されますが、Lambdaによる重複処理は起こり得ます。対象への反映は冪等にし、バッチ失敗時の再試行範囲と退避先を決めてください。さらに、24時間の保持期間内に滞留と失敗を検知し、退避したイベントを安全に再処理する運用まで用意して、初めて継続的に扱える構成になります。