--- title: "CSV一括インポートをAWS Lambda×SQSで実装｜S3イベント・部分失敗・冪等の設計" url: "https://www.issoh.co.jp/tech/details/1884/" published: 2024-01-26 updated: 2026-10-11 categories: ["AWS"] publisher: "株式会社一創" --- # CSV一括インポートをAWS Lambda×SQSで実装｜S3イベント・部分失敗・冪等の設計 数万行のCSVをS3に置き、Lambdaで読み込んでデータベースへ流し込む。この構成で本番の障害につながりやすいのは、処理ロジックよりもSQSとLambdaの上限値、リトライと重複の扱いです。ここではS3に置いたCSVを分割Lambda、SQS、ワーカーLambdaの2段構成で取り込む実装を、IAMとイベント通知の構築手順から、AWS公式ドキュメントのクォータ値、Python 3.13で実行したコードまで通しで組み立てます。 ## まとめ：CSV一括インポートで先に決める6つの設定 実装に入る前に確認しておきたいのは次の6点です。 - **処理時間と負荷に応じてLambdaを分割する**：通常のLambda実行環境の関数タイムアウト上限は900秒（15分）で、処理がこの上限に近づく場合や書き込み負荷を制御したい場合に分割します。読み取り役と書き込み役を分け、間にSQSを挟みます。 - **S3イベントのキーは必ずデコードする**：通知に載るオブジェクトキーはURLエンコード済みです。そのまま`get_object`に渡すと、空白や日本語を含むファイル名でキーが一致せず失敗します。 - **1メッセージは行数でなくバイト数で切る**：SQSの最大メッセージサイズは1 MiB。`SendMessageBatch`は10件までで、合計バイト数も1 MiBが上限です。 - **可視性タイムアウトは関数タイムアウトの6倍＋バッチウィンドウ**：AWSが公式に示す推奨値です。 - **部分失敗は`ReportBatchItemFailures`で返す**：これが無いと、1件の不正メッセージで同じバッチの全メッセージがキューに戻ります。 - **重複は分割側とワーカー側の両方で起きる前提で書く**：S3通知もSQSも「少なくとも1回」の配信です。書き込みは業務キーで上書きになる形にします。 ## 1本のLambdaでCSVを処理しきれなくなる境界 本記事は通常のLambda実行環境を前提とします。Lambda Managed Instancesでは、非同期呼び出しとSQSなどのイベントソースマッピングで最大5,400秒まで実行できるため、以下の900秒という制約とは区別します。[Lambdaのクォータ一覧](https://docs.aws.amazon.com/lambda/latest/dg/gettingstarted-limits.html)では、関数タイムアウトは900秒（15分）。メモリは128 MBから10,240 MBまで1 MB刻みで、1,769 MBで1 vCPU相当が割り当てられます。一時ディレクトリの`/tmp`は512 MBから10,240 MBです。 問題はメモリよりも時間です。1行の処理にデータベースへの書き込みや外部APIの呼び出しが入ると、1行10ミリ秒でも90,000行で900秒に達します。打ち切られた時点で途中までの行は書き込まれたまま残り、リトライすれば先頭から読み直しになります。 分割するかどうかは、1ファイルの処理時間の見積もりで決まります。行数×1行あたりの処理時間が900秒に対して十分小さく、書き込み先の負荷を絞る必要もないなら、1本のLambdaで足ります。SQSを挟む価値はありません。見積もりが900秒に近づく場合や、書き込み先への同時アクセス数を制御したい場合に、読み取りと処理を分けます。Lambdaそのものの課金体系や採用判断は[AWS Lambdaの仕組み・料金体系とコールドスタート対策](/tech/details/15407/)で扱っています。 ## S3からSQSまでの構成と接続設定 S3にCSVが置かれたイベントで分割Lambdaが起動し、CSVをストリームで読みながら一定量ずつSQSへ投入します。SQSのイベントソースマッピングがワーカーLambdaを呼び出し、ワーカーが書き込みを担当します。ここでは関数本体、同一リージョンの標準SQSキュー、DynamoDBテーブルが作成済みで、環境変数QUEUE\_URLとTABLE\_NAMEも設定済みとして、権限と接続の設定を示します。掲載コードは標準キュー用です。 ### IAMロール：分割Lambdaとワーカーの権限分離 2つの関数は触るリソースが違うので、実行ロールも分けます。次の例は分割Lambdaに足す業務側の権限で、CloudWatch Logsへの基本の書き込み権限は別に付けます。取り込み用プレフィックスの`s3:GetObject`とキューへの`sqs:SendMessage`です。キューをKMSのカスタマーマネージドキーで暗号化している場合は、送信側に`kms:GenerateDataKey`と`kms:Decrypt`、受信側に`kms:Decrypt`も要ります。`SendMessageBatch`専用のアクションはなく、`sqs:SendMessage`の許可でバッチ送信も通ります。 ``` { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "s3:GetObject", "Resource": "arn:aws:s3:::example-csv-bucket/incoming/*" }, { "Effect": "Allow", "Action": "sqs:SendMessage", "Resource": "arn:aws:sqs:ap-northeast-1:123456789012:csv-import" } ] } ``` ワーカー側には、AWSマネージドポリシーの[AWSLambdaSQSQueueExecutionRole](https://docs.aws.amazon.com/aws-managed-policy/latest/reference/AWSLambdaSQSQueueExecutionRole.html)を付けます。中身は`sqs:ReceiveMessage`、`sqs:DeleteMessage`、`sqs:GetQueueAttributes`とCloudWatch Logsへの書き込み3つです。ただしすべてのアクションでリソースが`*`なので、本番ではSQSの3つのアクションを対象キューのARNに絞り、Logsのアクションは関数のロググループに絞った別のステートメントに分けるほうが安全です。書き込み先がDynamoDBなら、これに対象テーブルへの`dynamodb:BatchWriteItem`を足します。 ### S3イベント通知：prefixとsuffixによる対象の限定 S3からLambdaを呼ぶには、関数側のリソースベースポリシーで`s3.amazonaws.com`に呼び出しを許可します。コンソールで通知を設定すればこの許可は自動で入りますが、CLIやIaCで組むときは自分で追加します（[S3の公式ガイド](https://docs.aws.amazon.com/AmazonS3/latest/userguide/grant-destinations-permissions-to-s3.html)）。続けて通知設定とイベントソースマッピングを作ります。なお`put-bucket-notification-configuration`は通知設定の追加ではなく全体の置き換えです。次の例は通知が未設定の新規バケット向けで、既にSNSやSQSへの通知があるバケットでは、現在の設定を取得して統合したJSONを渡してください。 ``` # 1. S3から分割Lambdaを呼べるようにする（リソースベースポリシー） aws lambda add-permission \ --function-name csv-import-splitter \ --statement-id s3-invoke \ --action lambda:InvokeFunction \ --principal s3.amazonaws.com \ --source-arn arn:aws:s3:::example-csv-bucket \ --source-account 123456789012 # 2. incoming/配下の.csvだけを通知対象にする aws s3api put-bucket-notification-configuration \ --bucket example-csv-bucket \ --notification-configuration '{ "LambdaFunctionConfigurations": [{ "LambdaFunctionArn": "arn:aws:lambda:ap-northeast-1:123456789012:function:csv-import-splitter", "Events": ["s3:ObjectCreated:*"], "Filter": {"Key": {"FilterRules": [ {"Name": "prefix", "Value": "incoming/"}, {"Name": "suffix", "Value": ".csv"} ]}} }] }' # 3. キューとワーカーをつなぐ（部分失敗の報告と同時実行の上限を最初から入れる） aws lambda create-event-source-mapping \ --function-name csv-import-worker \ --event-source-arn arn:aws:sqs:ap-northeast-1:123456789012:csv-import \ --batch-size 10 \ --function-response-types ReportBatchItemFailures \ --scaling-config '{"MaximumConcurrency": 10}' ``` 通知を`incoming/`配下の`.csv`に絞っているのは、取り込み結果やエラーレポートを同じバケットに書き出したときに、それがまた分割Lambdaを起動しないようにするためです。`--scaling-config`の最大同時実行数は2から1,000の範囲で指定します。値の決め方は後述の同時実行数の節で扱います。バケットの作成や権限設計の基本は[AWS S3の使い方｜CLIでのバケット作成から権限設計・署名付きURLまでの実装手順](/tech/details/17489/)にまとめています。 ## 分割Lambdaの実装：URLデコードとバイト予算 「1メッセージ500行」のような固定行数で切る実装をよく見かけますが、列数の多いCSVで破綻します。1 MiBはバイト数の上限だからです。行をJSONにエンコードした実バイト数を積算し、予算に達したら切ります。通常の行のまとまりは240 KiBを目安に分割します。ただし、この予算には外側のJSONやファイルキーを含めておらず、単独で予算を超える行も許可するため、メッセージ本文の厳密な240 KiB上限ではありません。さらにDynamoDBでは属性名と値を含む1アイテムが400 KB以内である必要があり、SQSに送れる行でも書き込めない場合があります。 ``` import csv import io import json import os from urllib.parse import unquote_plus import boto3 s3 = boto3.client("s3") sqs = boto3.client("sqs") QUEUE_URL = os.environ["QUEUE_URL"] MAX_BODY_BYTES = 240 * 1024 MAX_BATCH_BYTES = 1000 * 1024 def pack_rows(reader, max_bytes): rows, size = [], 2 for row in reader: encoded = len(json.dumps(row, ensure_ascii=False).encode("utf-8")) + 2 if rows and size + encoded > max_bytes: yield rows rows, size = [], 2 rows.append(row) size += encoded if rows: yield rows def lambda_handler(event, context): sent = 0 for record in event["Records"]: bucket = record["s3"]["bucket"]["name"] key = unquote_plus(record["s3"]["object"]["key"]) sent += split_object(bucket, key) return {"messages": sent} def split_object(bucket, key): body = s3.get_object(Bucket=bucket, Key=key)["Body"] reader = csv.DictReader(io.TextIOWrapper(body, encoding="utf-8-sig", newline="")) entries, batch_bytes, sent = [], 0, 0 for rows in pack_rows(reader, MAX_BODY_BYTES): message = json.dumps({"key": key, "rows": rows}, ensure_ascii=False) size = len(message.encode("utf-8")) if size > MAX_BATCH_BYTES: raise ValueError(f"{key}: 1行だけで{size}バイトあり、SQSに載りません") if entries and (len(entries) == 10 or batch_bytes + size > MAX_BATCH_BYTES): sent += flush(entries) entries, batch_bytes = [], 0 entries.append({"Id": str(len(entries)), "MessageBody": message}) batch_bytes += size if entries: sent += flush(entries) return sent def flush(entries): response = sqs.send_message_batch(QueueUrl=QUEUE_URL, Entries=entries) if response.get("Failed"): raise RuntimeError(response["Failed"]) return len(entries) ``` ファイル全体をメモリに載せない点も意図的です。`get_object`が返す`Body`はストリームで、`TextIOWrapper`と`csv.DictReader`を重ねると1行ずつ読み進めます。ここで`read()`を呼んで文字列化すると、CSVのサイズがそのまま関数のメモリ消費になります。`newline=""`は[csvモジュールの公式ドキュメント](https://docs.python.org/3.13/library/csv.html)が求める指定で、これが無いと引用符で囲んだセル内の改行CRLFがLFに変わります。1行だけで1 MiBを超える行はバイト予算で切っても載らないので、送信前に`ValueError`で止めています。各セル12万文字×9列の行で試すと、1,080,131バイトでこのエラーになりました。ただし、CSVパーサーにもセル長の制限があります。掲載コードをPython 3.13.14で実行すると、単一セル14万文字の入力はSQS送信前にcsv.Errorで停止します。長いセルを許可する場合は、csv.field\_size\_limitで業務上の上限を設定します。 ### S3イベントのオブジェクトキーとURLデコード コードの`unquote_plus`は飾りではありません。[S3のイベントメッセージ構造](https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-content-structure.html)には、オブジェクトキーがURLエンコードされ、`red flower.jpg`は`red+flower.jpg`になると明記されています。AWSのS3トリガーのチュートリアルも`unquote_plus`でデコードしています。 `incoming/2026年10月 受注.csv`というキーは、通知では`incoming/2026%E5%B9%B410%E6%9C%88+%E5%8F%97%E6%B3%A8.csv`になります。これをそのまま`get_object`に渡すと、存在しないキーを読みにいくことになります。返るエラーは権限で変わり、[GetObjectのリファレンス](https://docs.aws.amazon.com/botocore/latest/reference/services/s3/client/get%5Fobject.html)どおり、バケットへの`s3:ListBucket`があれば404（`NoSuchKey`）、無ければ403（`AccessDenied`）です。前掲のIAM例はListBucketを付けていないので403になり、権限の設定ミスと取り違えやすくなります。英数字のファイル名だけでテストしていると本番まで気づきません。 ### SendMessageBatchの二重上限と実行結果 送信をまとめる条件が2つあるのは、[SendMessageBatchのAPIリファレンス](https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API%5FSendMessageBatch.html)が件数と合計サイズの両方に上限を置いているからです。件数は1リクエスト10件まで、メッセージ単体とバッチ合計のバイト数はどちらも1 MiB（1,048,576バイト）です。超えると`TooManyEntriesInBatchRequest`または`BatchRequestTooLong`が返ります。SQSの上限は[2025年8月4日に256 KiBから1 MiBへ引き上げられた](https://aws.amazon.com/about-aws/whats-new/2025/08/amazon-sqs-max-payload-size-1mib/)ので、「上限が256 KiB」と書かれた古い記事は仕様として読み替えが必要です。掲載コードが240 KiBに抑えているのは設計上の選択で、1件の失敗で再処理される行数を減らすためです。 このコードをPython 3.13.14とboto3 1.43.111で、スタブしたS3とSQSに対して掲載どおり実行した結果が次のとおりです。入力は10列20,000行、2,559,115バイトのCSVで、キーは上の日本語ファイル名をエンコードした形で渡しています。スタブ側では1リクエスト10件・合計1 MiB・Idの重複の3つを検査しています。 | 項目 | 実行結果 | | ----------------------- | --------------- | | 生成メッセージ数 | 18 | | SendMessageBatch 呼び出し回数 | 5 | | 1リクエストあたりの件数 | 4（最後のみ2） | | 1リクエストの合計バイト数 | 448,924〜982,864 | | 1メッセージあたりの行数 | 927〜1,130 | | メッセージ本文のバイト数 | 203,231〜245,790 | 行数が927行から1,130行まで振れているのは、バイト予算で切っているためです。同じCSVを「1メッセージ1行」で送ると20,000メッセージ、`SendMessageBatch`は2,000回になります。API呼び出しが400倍になるうえ、ワーカーの起動回数も跳ね上がります。 ## ワーカーLambdaの実装：部分失敗を返す書き込み処理 ワーカーは1メッセージ内の行をまとめて書き込み、失敗したメッセージIDだけを返します。前提は、環境変数`TABLE_NAME`で渡すDynamoDBテーブルのパーティションキーが文字列型の`pk`で、CSVに空でない`id`列があり、ヘッダーの重複や列数の不一致がないことです。掲載コードはこれらを検証しないため、本番では読み込み時に検証し、不正行を行番号付きで分離します。例外をそのまま外へ投げてはいけません。後述のとおり、バッチ全体の失敗として扱われます。 ``` import json import os import boto3 table = boto3.resource("dynamodb").Table(os.environ["TABLE_NAME"]) def lambda_handler(event, context): failures = [] for record in event["Records"]: try: payload = json.loads(record["body"]) with table.batch_writer(overwrite_by_pkeys=["pk"]) as writer: for row in payload["rows"]: writer.put_item(Item={**row, "pk": payload["key"] + "#" + row["id"]}) except Exception as exc: print(f"failed messageId={record['messageId']} error={type(exc).__name__}") failures.append({"itemIdentifier": record["messageId"]}) return {"batchItemFailures": failures} ``` 生成した`pk`を辞書展開の後ろに置いているのは、CSVに`pk`という列があったときに主キーを上書きされないためです。逆順にすると、CSVの`pk`列の値がそのまま主キーになり、別の行を上書きします。失敗時のログはメッセージIDと例外の型だけにし、本文の個人情報がログに残らないようにしています。 `overwrite_by_pkeys`を付けているのは、1メッセージの中に同じ業務キーの行が2回出てくるCSVへの備えです。[BatchWriteItemのAPIリファレンス](https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API%5FBatchWriteItem.html)は、同じキーのアイテムを2つ含むリクエストをまるごと拒否すると定めています。拒否されたメッセージは何度リトライしても通らず、最後はDLQ行きです。この指定があると`batch_writer`がバッファ内の重複を後勝ちで1件にまとめます。スタブで確かめると、指定なしでは3件がそのまま送られ、指定ありでは後から来た値の2件に減りました。 スタブしたテーブルに、正常なメッセージ2件と本文が壊れたJSON 1件を渡した戻り値は`{"batchItemFailures": [{"itemIdentifier": "id2"}]}`で、正常な2件分の2,043行は書き込まれました。壊れた1件だけがキューに戻ります。DynamoDBの書き込み単価と請求が膨らむ設計は[DynamoDBの料金を東京リージョンの実額で試算する](/tech/details/17798/)で扱っています。 ## イベントソースマッピングの設定値と根拠 ### バッチサイズとバッチウィンドウ：10を超えるなら1秒以上が必須 バッチサイズの既定値は10です。標準キューでは最大10,000件まで上げられますが、10を超える値には`MaximumBatchingWindowInSeconds`を1秒以上設定する必要があります。FIFOキューの上限は10です。バッチウィンドウの最大値は5分です。 設定値どおりの件数が届くとは限りません。Lambdaは同期呼び出しのペイロード上限6 MBを超えない範囲でしかレコードを詰めず、レコードごとのメタデータもこの6 MBに数えられます。前掲の試験入力では1メッセージが最大245,790バイトだったので、同じ程度の大きさのメッセージなら、1回の呼び出しに入るのは6 MBに対して20件台までです。この入力でバッチサイズを100にしても、100件はまとまって届きません。 低トラフィックのキューでバッチウィンドウを使うと、設定値が20秒未満でもLambdaが最大20秒待つことがあります（[LambdaとSQSの公式ガイド](https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html)）。取り込みの応答性を秒単位で詰めたいなら、バッチウィンドウは0のままにします。 ### 可視性タイムアウト：関数タイムアウトの6倍＋バッチウィンドウ 可視性タイムアウトの既定値は30秒、最大は12時間です。AWSはソースキューの可視性タイムアウトを関数タイムアウトの6倍以上にし、バッチウィンドウを使うならその秒数を足すよう求めています。6倍の余裕は、前のバッチの処理中に関数がスロットリングされたときのリトライ分です。 | 関数タイムアウト | バッチウィンドウ | 可視性タイムアウトの推奨最小値 | | -------- | -------- | --------------- | | 30秒 | 0秒 | 180秒 | | 60秒 | 0秒 | 360秒 | | 60秒 | 30秒 | 390秒 | | 300秒 | 60秒 | 1,860秒 | 逆方向の制約もあります。関数タイムアウトがキューの可視性タイムアウトを超える設定はイベントソースマッピングの作成時・更新時に検証され、エラーになります。関数タイムアウトを伸ばすときは、先にキュー側を広げてください。可視性タイムアウトそのものの仕組みは[Amazon SQSとは？標準キューとFIFOの違い・可視性タイムアウトとDLQ設計](/tech/details/17241/)で解説しています。 ### 同時実行数：MaximumConcurrencyによる上限設定 [SQSイベントソースのスケーリング](https://docs.aws.amazon.com/lambda/latest/dg/services-sqs-scaling.html)は、メッセージが溜まるとまず5バッチを5つの同時実行で処理し、残っていれば毎分最大300ずつ同時実行を足して、最大1,250に達します。ただしアカウントの同時実行クォータが既定の1,000のままなら、上限はその1,000です。 ここに落とし穴があります。この1,000はリージョン内の全関数で共有する枠です。最大同時実行数を設定しないまま数十万行のCSVを流すと、ワーカーが枠を使い切り、分割Lambdaや無関係なAPIの関数までスロットリングされます。書き込み先がRDSのような接続数の限られたデータベースなら、接続数は同時実行数に1実行環境あたりの接続数を掛けた分まで増えます。だから構築手順では`MaximumConcurrency`を最初から入れています。決め方は「書き込み先が受けられる同時接続数÷1実行環境あたりの接続数」で逆算し、2〜1,000の範囲で指定します。関数の予約済み同時実行数を使う場合は、それより大きい値にしないでください。 立ち上がりを速くしたい場合はプロビジョンドモード（毎分最大1,000の同時実行を追加、最大100,000、追加料金あり）がありますが、最大同時実行数の設定とは併用できません。 ## 部分失敗とデッドレターキューの設計 ### バッチ全体が失敗扱いになる5つの返し方 先に粒度を確認しておきます。キューへ戻せる最小単位はメッセージであって、行ではありません。1メッセージに1,000行を詰めた設計なら、1行の不正で1,000行が再処理されます。 部分失敗は、イベントソースマッピングの`FunctionResponseTypes`に`ReportBatchItemFailures`を指定して有効にします。設定したのに失敗メッセージが戻ってこない、という詰まり方をするのがここです。[SQSイベントソースのエラー処理](https://docs.aws.amazon.com/lambda/latest/dg/services-sqs-errorhandling.html)によると、Lambdaは次のいずれかを返されると部分失敗ではなくバッチ全体の失敗として扱います。 - 無効なJSONレスポンス - 空文字列の`itemIdentifier` - nullの`itemIdentifier` - キー名を誤った`itemIdentifier` - 存在しないメッセージIDを指す`itemIdentifier` 関数が例外を送出した場合も全体失敗です。だからワーカー側で例外をすべて捕捉し、失敗したメッセージIDを詰めて返す形にしています。逆に空のリスト、nullのリスト、空またはnullの`EventResponse`は、いずれも完全成功として扱われます。 FIFOキューでは作法が違います。最初の失敗で処理を止め、失敗したメッセージと未処理のメッセージをすべて`batchItemFailures`に入れて返します。そうしないと順序が崩れます。この返し方を手書きしない選択肢として、AWSは[Powertools for AWS LambdaのBatch Processor](https://docs.aws.amazon.com/powertools/python/latest/utilities/batch/)が部分失敗のロジックを自動で処理すると案内しています。 ### デッドレターキューとmaxReceiveCount 何度リトライしても通らないメッセージは、デッドレターキュー（DLQ）へ退避させます。AWSはソースキューのリドライブポリシーで`maxReceiveCount`を5以上にするよう推奨しています。3などの小さい値にすると、関数の不具合ではなく一時的なスロットリングだけでメッセージがDLQへ移ります。 保持期間の勘定にも落とし穴があります。標準キューでは、メッセージがDLQへ移ってもエンキュー時刻がリセットされません。元のキューで1日滞留したメッセージは、DLQの保持期間が既定の4日なら残り3日で消えます。DLQの保持期間は元のキューより長くしてください。DLQをEventBridge経由の構成のどこに置くかは[EventBridgeとSQS・Lambdaの非同期構成](/tech/details/6034/)で比較しています。 ## 同じ行が2回届く前提の書き込み設計 重複はワーカーの手前だけで起きるわけではありません。経路上に重複の発生源が3つあります。 - **S3イベント通知**：[S3の通知](https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-how-to-event-types-and-destinations.html)は少なくとも1回の配信で、再試行の仕組みにより同じオブジェクトイベントの通知がまれに重複します。 - **分割Lambdaの自動リトライ**：S3はLambdaを非同期で呼び出します。[非同期呼び出し](https://docs.aws.amazon.com/lambda/latest/dg/invocation-async-error-handling.html)で関数がエラーを返すと、既定でさらに2回（1分後と2分後）再実行されます。途中まで送信してから落ちた分割Lambdaは、再実行で先頭のメッセージをもう一度送ります。 - **SQSイベントソースマッピング**：AWS自身が「レコードの重複処理が起こり得る」と警告し、関数を冪等にするよう強く推奨しています。 3つとも送る側では防げないので、受ける側で吸収します。行から決まるキーで上書きになる書き込みにするのが最も簡単です。前掲のワーカーはファイルキーと行の`id`列を連結した`pk`で`put_item`しており、同じメッセージが何回届いても結果は変わりません。リレーショナルデータベースなら、業務上の一意キーに対するUPSERTが同じ役割を果たします。 ただし、これで保証されるのは「同じ行が何度届いても同じ結果になる」ことまでです。同じ業務キーで値の違う行が別々のメッセージに入っていると、ワーカーの並列実行で到着順が入れ替わり、どちらの値が残るかは決まりません。CSVが更新履歴を含むなら、更新日時やバージョン列を比べる条件付き書き込みにしてください。また、この実装は通知の`versionId`を使わず最新版を読むので、処理中に同じキーへ別のファイルを上書きアップロードしない運用が前提です。上書きがあり得るなら、`versionId`を引き回し、`s3:GetObjectVersion`権限を付与したうえで、`get_object`の`VersionId`引数に指定して読みます。 キーの作り方を間違えると効果が消えます。ファイル名と行番号を組み合わせたキーは、同じデータを別のファイル名で再アップロードしたときに別レコードとして入ります。注文番号や商品コードのような業務キーが使えるなら、そちらを優先してください。 ## 詰まりと取りこぼしを検知する3つのメトリクスと件数照合 部分失敗の実装が壊れていることは、コードを読んでも気づきにくく、CloudWatchのSQSメトリクスに先に出ます。`NumberOfMessagesDeleted`はキューから削除されたメッセージ数で、これが0に落ちていれば関数のレスポンスが失敗メッセージを正しく返せていない兆候です。`ApproximateAgeOfOldestMessage`はキュー内で最も古いメッセージの滞留時間で、急上昇は同じ原因か、処理が投入に追いついていないことを示します。 ただし`ApproximateAgeOfOldestMessage`には前提があります。標準キューで`maxReceiveCount`を3より大きくしていると、3回以上受信されて削除されなかったメッセージをSQSがキュー末尾へ回し、このメトリクスは閾値を超えていない次のメッセージの経過時間を示すようになります。前節の推奨どおり5にしている場合、このメトリクスだけでは毒メッセージの滞留を見落とします。取り込みバッチの基本セットは、この2つにDLQ側の`ApproximateNumberOfMessagesVisible`を足した3本のアラームです。ただし、ワーカーが例外を握りつぶして成功扱いで返した行は、キューにもDLQにも残らないのでこの3本では見つかりません。掲載の分割Lambdaが返すのは送信メッセージ数だけです。取りこぼしの検知には、入力全体から求めた書き込み予定キーと内容の照合情報を別途保存し、書き込み先と比較する処理が必要です。行数だけの比較では、重複キーの上書きや同件数の内容不一致を区別できません。 ## Step Functions分散マップとの使い分け S3上のCSVを並列処理する方法として、AWSは[Step Functionsの分散マップ](https://docs.aws.amazon.com/step-functions/latest/dg/state-map-distributed.html)も用意しています。`ItemReader`がS3のCSVを直接読み、行ごと（または`ItemBatcher`でまとめた単位ごと）に子ワークフローを起動する仕組みで、分割Lambdaが要りません。どちらを選ぶかは、次の差で決まります。 | 観点 | 分割Lambda＋SQS | Step Functions分散マップ | | -------- | -------------------- | ------------------------ | | CSVの読み取り | 分割Lambda（900秒以内） | ItemReader（1ファイル10 GBまで） | | 1単位の大きさ | 1メッセージ1 MiBまで | 1バッチ256 KiBまで | | 並列度の既定 | 最大1,250（アカウント枠内） | 子ワークフロー10,000 | | 一部失敗時の動作 | 部分失敗の設定で失敗分のみ再試行 | 既定は1件の失敗でMap Run全体が失敗 | | 列数が合わない行 | 不足はNone、超過はキーNoneに残る | 不足は空文字、超過は黙って捨てる | | 課金の単位 | Lambda実行＋SQSリクエスト | 子の起動1件ごとの遷移＋子の実行料金 | 分散マップの強みは、分割Lambdaの900秒の壁が消えることです。[ItemReaderの仕様](https://docs.aws.amazon.com/step-functions/latest/dg/input-output-itemreader.html)ではヘッダーを先頭行から取る`FIRST_ROW`と定義側で与える`GIVEN`を選べ、区切り文字もカンマ以外に切り替えられます。 一方で、データ品質の扱いはSQS構成より荒くなります。`ToleratedFailurePercentage`の既定値は0で、1件でも子ワークフローが失敗するとMap Run全体が失敗します。不正行が混ざっても処理を続ける場合は、子ワークフローで入力エラーを記録して正常終了させるか、ToleratedFailureCountまたはToleratedFailurePercentageで許容する失敗を設定します。もう1つ見落とされやすいのが、ヘッダーより列が多い行の超過フィールドを黙って捨てる仕様です。エラーにならないので、区切り文字のずれたCSVが欠けたまま取り込まれます。Pythonの`csv.DictReader`は超過分をキー`None`のリストに残すため、分割Lambdaなら`None in row`で検出して弾けます。全フィールドが文字列として渡るため、数値や日付への型変換は別途必要です。この点は、既定設定のPythonのcsv.DictReaderを使う掲載コードでも同じです。 判断としては、ファイルが分割Lambdaの900秒に収まらないほど大きいか、取り込み全体の成否を1つの実行単位で管理したいなら分散マップです。行単位のエラーを隔離しながら流し続けたい、既存のSQSワーカーや監視に乗せたいなら、ここまでのSQS構成が向きます。 ## Lambda＋SQSによるCSV取り込みの不採用条件 次の3つの条件では、SQSを挟まない選択肢のほうが素直です。 数千行かつ単純変換のCSVなら、SQSは不要です。5つの構成要素を運用する対価に見合わないので、Lambda1本でS3から読んで書き込みます。 厳密な取り込み順序が必要な場合は、FIFOキューを選ぶ前に考え直す価値があります。FIFOはバッチサイズ10が上限で、最初の失敗以降も未処理として返す必要があるため、1件の不正行が後続を止めます。行順を保ちたいだけなら、行番号をレコードに持たせて標準キューで取り込み、読む側で並べ替えます。 1ファイルが分割Lambdaの900秒で読み切れない大きさになる、または書き込み先の制約で同時実行数を絞った結果、取り込みが許容時間に収まらないと試算で分かったなら、Lambdaの枠に収める設計にこだわる必要はありません。[AWS GlueのETLジョブ](/tech/details/17472/)、Step Functionsの分散マップ、あるいはデータベース側の一括ロード機能に寄せてください。どの構成が自社のデータ量と運用体制に合うか判断がつかない場合は、[AWSを含むクラウドのインフラ構築支援](/service/system/aws/)で設計段階から相談できます。 ## よくある質問 ### SQSのメッセージサイズ上限は？1 MiBを超えるデータはどう渡しますか？ 最小1バイト、最大1,048,576バイト（1 MiB）です。超えるペイロードは、Amazon SQS Extended Client Library（JavaとPython向け）で実データをS3に置き、メッセージにはS3への参照だけを載せます。最大ペイロードサイズは2 GBです。SNSの既定値は256 KiBですが、対応する配信先などの条件を満たすトピックでは、MaximumMessageSizeを設定して最大1 MiBまで拡張できます。サービスごとの上限の違いは[AWS SQS・SNS・SESの違いと使い分け](/tech/details/9310/)で比較しています。 ### 可視性タイムアウトは何秒に設定すればよいですか？ 関数タイムアウトの6倍以上で、バッチウィンドウを使うならその秒数を足した値です。関数タイムアウト60秒、バッチウィンドウ30秒なら390秒になります。 ### S3 SelectでCSVを絞ってからLambdaに渡せますか？ 新しく始める場合は使えません。AWS公式ドキュメントの冒頭に「Amazon S3 Select is no longer available to new customers.」と記載されており、使い続けられるのは既存の利用者だけです。取り込み時のフィルタで済むなら分割Lambdaで不要行を捨て、アドホックな分析が目的なら[Amazon Athena](/tech/details/17734/)を使います。 ### SQSのコンシューマーをLambda以外で実装する場合の選択肢は？ 常時起動のワーカープロセスからポーリングする方式があります。処理が15分を超える、あるいはコネクションプールを維持したい場合はこちらが向きます。RubyならActive JobのバックエンドとしてSQSを扱うShoryukenがあり、設定とリトライ設計は[Shoryuken（Ruby gem）とは？SQSワーカーの設定とActive Job連携](/tech/details/3429/)で解説しています。 ### Shift\_JISのCSVを取り込むにはどうすればよいですか？ 分割Lambdaの`encoding`を、出力元の実際の文字コードに合わせて変えます。Windowsのアプリケーションが書き出した「Shift\_JIS」のCSVは、Microsoftの拡張文字を含むcp932であることが多く、Pythonでは`shift_jis`より`cp932`を指定するほうが通ります。丸数字の1、全角チルダ、平方メートルの記号は`shift_jis`では扱えず、`cp932`なら扱えます。ただし平方ミリの記号はcp932にも無いので、こうした文字を含むデータはUTF-8で受け取る必要があります。文字コードが混在する運用なら、アップロード側でUTF-8に統一させるほうが安全です。 ## 関連記事 - [Amazon SQSとは？標準キューとFIFOの違い・可視性タイムアウトとDLQ設計を解説](/tech/details/17241/) - [EventBridgeとSQS・Lambdaの非同期構成｜キューポリシー・部分バッチ応答・DLQの置き場所](/tech/details/6034/) - [AWS Lambdaとは？仕組み・料金体系とコールドスタート対策・採用判断を実装者目線で解説](/tech/details/15407/) - [AWS S3の使い方｜CLIでのバケット作成から権限設計・署名付きURLまでの実装手順](/tech/details/17489/) - [AWS GlueでETLジョブを動かす手順｜PySparkスクリプトとDPU課金の設計](/tech/details/17472/) --- 出典: [CSV一括インポートをAWS Lambda×SQSで実装｜S3イベント・部分失敗・冪等の設計]()（株式会社一創）