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のクォータ一覧では、関数タイムアウトは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の仕組み・料金体系とコールドスタート対策で扱っています。

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を付けます。中身は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の公式ガイド）。続けて通知設定とイベントソースマッピングを作ります。なお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までの実装手順にまとめています。

分割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モジュールの公式ドキュメントが求める指定で、これが無いと引用符で囲んだセル内の改行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のイベントメッセージ構造には、オブジェクトキーが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のリファレンスどおり、バケットへのs3:ListBucketがあれば404（NoSuchKey）、無ければ403（AccessDenied）です。前掲のIAM例はListBucketを付けていないので403になり、権限の設定ミスと取り違えやすくなります。英数字のファイル名だけでテストしていると本番まで気づきません。

SendMessageBatchの二重上限と実行結果

送信をまとめる条件が2つあるのは、SendMessageBatchのAPIリファレンスが件数と合計サイズの両方に上限を置いているからです。件数は1リクエスト10件まで、メッセージ単体とバッチ合計のバイト数はどちらも1 MiB（1,048,576バイト）です。超えるとTooManyEntriesInBatchRequestまたはBatchRequestTooLongが返ります。SQSの上限は2025年8月4日に256 KiBから1 MiBへ引き上げられたので、「上限が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つを検査しています。

行数が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リファレンスは、同じキーのアイテムを2つ含むリクエストをまるごと拒否すると定めています。拒否されたメッセージは何度リトライしても通らず、最後はDLQ行きです。この指定があるとbatch_writerがバッファ内の重複を後勝ちで1件にまとめます。スタブで確かめると、指定なしでは3件がそのまま送られ、指定ありでは後から来た値の2件に減りました。

スタブしたテーブルに、正常なメッセージ2件と本文が壊れたJSON 1件を渡した戻り値は{"batchItemFailures": [{"itemIdentifier": "id2"}]}で、正常な2件分の2,043行は書き込まれました。壊れた1件だけがキューに戻ります。DynamoDBの書き込み単価と請求が膨らむ設計はDynamoDBの料金を東京リージョンの実額で試算するで扱っています。

イベントソースマッピングの設定値と根拠

バッチサイズとバッチウィンドウ：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の公式ガイド）。取り込みの応答性を秒単位で詰めたいなら、バッチウィンドウは0のままにします。

可視性タイムアウト：関数タイムアウトの6倍＋バッチウィンドウ

可視性タイムアウトの既定値は30秒、最大は12時間です。AWSはソースキューの可視性タイムアウトを関数タイムアウトの6倍以上にし、バッチウィンドウを使うならその秒数を足すよう求めています。6倍の余裕は、前のバッチの処理中に関数がスロットリングされたときのリトライ分です。

逆方向の制約もあります。関数タイムアウトがキューの可視性タイムアウトを超える設定はイベントソースマッピングの作成時・更新時に検証され、エラーになります。関数タイムアウトを伸ばすときは、先にキュー側を広げてください。可視性タイムアウトそのものの仕組みはAmazon SQSとは？標準キューとFIFOの違い・可視性タイムアウトとDLQ設計で解説しています。

同時実行数：MaximumConcurrencyによる上限設定

SQSイベントソースのスケーリングは、メッセージが溜まるとまず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イベントソースのエラー処理によると、Lambdaは次のいずれかを返されると部分失敗ではなくバッチ全体の失敗として扱います。

• 無効なJSONレスポンス
• 空文字列のitemIdentifier
• nullのitemIdentifier
• キー名を誤ったitemIdentifier
• 存在しないメッセージIDを指すitemIdentifier

関数が例外を送出した場合も全体失敗です。だからワーカー側で例外をすべて捕捉し、失敗したメッセージIDを詰めて返す形にしています。逆に空のリスト、nullのリスト、空またはnullのEventResponseは、いずれも完全成功として扱われます。

FIFOキューでは作法が違います。最初の失敗で処理を止め、失敗したメッセージと未処理のメッセージをすべてbatchItemFailuresに入れて返します。そうしないと順序が崩れます。この返し方を手書きしない選択肢として、AWSはPowertools for AWS LambdaのBatch Processorが部分失敗のロジックを自動で処理すると案内しています。

デッドレターキューとmaxReceiveCount

何度リトライしても通らないメッセージは、デッドレターキュー（DLQ）へ退避させます。AWSはソースキューのリドライブポリシーでmaxReceiveCountを5以上にするよう推奨しています。3などの小さい値にすると、関数の不具合ではなく一時的なスロットリングだけでメッセージがDLQへ移ります。

保持期間の勘定にも落とし穴があります。標準キューでは、メッセージがDLQへ移ってもエンキュー時刻がリセットされません。元のキューで1日滞留したメッセージは、DLQの保持期間が既定の4日なら残り3日で消えます。DLQの保持期間は元のキューより長くしてください。DLQをEventBridge経由の構成のどこに置くかはEventBridgeとSQS・Lambdaの非同期構成で比較しています。

同じ行が2回届く前提の書き込み設計

重複はワーカーの手前だけで起きるわけではありません。経路上に重複の発生源が3つあります。

• S3イベント通知：S3の通知は少なくとも1回の配信で、再試行の仕組みにより同じオブジェクトイベントの通知がまれに重複します。
• 分割Lambdaの自動リトライ：S3はLambdaを非同期で呼び出します。非同期呼び出しで関数がエラーを返すと、既定でさらに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の分散マップも用意しています。ItemReaderがS3のCSVを直接読み、行ごと（またはItemBatcherでまとめた単位ごと）に子ワークフローを起動する仕組みで、分割Lambdaが要りません。どちらを選ぶかは、次の差で決まります。

分散マップの強みは、分割Lambdaの900秒の壁が消えることです。ItemReaderの仕様ではヘッダーを先頭行から取る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ジョブ、Step Functionsの分散マップ、あるいはデータベース側の一括ロード機能に寄せてください。どの構成が自社のデータ量と運用体制に合うか判断がつかない場合は、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の違いと使い分けで比較しています。

可視性タイムアウトは何秒に設定すればよいですか？

関数タイムアウトの6倍以上で、バッチウィンドウを使うならその秒数を足した値です。関数タイムアウト60秒、バッチウィンドウ30秒なら390秒になります。

S3 SelectでCSVを絞ってからLambdaに渡せますか？

新しく始める場合は使えません。AWS公式ドキュメントの冒頭に「Amazon S3 Select is no longer available to new customers.」と記載されており、使い続けられるのは既存の利用者だけです。取り込み時のフィルタで済むなら分割Lambdaで不要行を捨て、アドホックな分析が目的ならAmazon Athenaを使います。

SQSのコンシューマーをLambda以外で実装する場合の選択肢は？

常時起動のワーカープロセスからポーリングする方式があります。処理が15分を超える、あるいはコネクションプールを維持したい場合はこちらが向きます。RubyならActive JobのバックエンドとしてSQSを扱うShoryukenがあり、設定とリトライ設計はShoryuken（Ruby gem）とは？SQSワーカーの設定とActive Job連携で解説しています。

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設計を解説
• EventBridgeとSQS・Lambdaの非同期構成｜キューポリシー・部分バッチ応答・DLQの置き場所
• AWS Lambdaとは？仕組み・料金体系とコールドスタート対策・採用判断を実装者目線で解説
• AWS S3の使い方｜CLIでのバケット作成から権限設計・署名付きURLまでの実装手順
• AWS GlueでETLジョブを動かす手順｜PySparkスクリプトとDPU課金の設計