View a markdown version of this page

耐久関数を使用したイベントソースマッピング - AWS Lambda

耐久関数を使用したイベントソースマッピング

耐久関数は、すべての Lambda イベントソースマッピングで機能します。耐久関数のイベントソースマッピングは、標準関数の設定と同じ方法で設定します。イベントソースマッピングは、Amazon SQS、Kinesis、DynamoDB Streams などのイベントソースを自動的にポーリングし、レコードのバッチで関数を呼び出します。

イベントソースマッピングは、複雑なマルチステップワークフローでストリームまたはキューを処理する耐久関数に有用です。例えば、再試行、外部 API コール、人間の承認を使用して Amazon SQS メッセージを処理する耐久関数を作成できます。

イベントソースマッピングが耐久関数を呼び出す方法

イベントソースマッピングは、耐久関数を同期的に呼び出し、耐久性のある実行が完全に終了するのを待ってから、次のバッチを処理したり、レコードを処理済みとしてマークします。合計耐久実行時間が該当するタイムアウト制限 (デフォルトでは 15 分、Lambda マネージドインスタンスで実行されている関数では最大 90 分) を超える場合、実行はタイムアウトして失敗します。イベントソースマッピングはタイムアウト例外を受け取り、再試行設定に従って処理します。

実行期間の制限

耐久性のある関数がイベントソースマッピングによって呼び出される場合、合計耐久実行時間は最大関数タイムアウトを超えることはできません。最大関数タイムアウトは、デフォルトの (オンデマンド) キャパシティモードで実行されている関数の場合は 15 分、Lambda マネージドインスタンス で実行されている関数の場合は最大 90 分です (Amazon MQ および Amazon DocumentDB イベントソースマッピングを除き、これらは 15 分に制限されたままです)。この制限は、個々の関数呼び出しだけでなく、最初から最後までの耐久性のある実行全体に適用されます。

この制限は、呼び出しごとの Lambda 関数タイムアウトとは別ですが、どちらも同じ最大値を共有します (デフォルトでは 15 分、非同期およびイベントソースマッピング呼び出しの場合は Lambda マネージドインスタンスで最大 90 分)。関数タイムアウトは、個々の呼び出しを実行できる時間を制御し、耐久性のある実行のタイムアウトは、実行開始から完了までの合計経過時間を制御します。

シナリオ例:

  • 有効 (任意のキャパシティモード): 耐久関数は、Amazon SQS メッセージを 3 つの手順で処理します。各手順には 2 分かかり、最後の手順を完了するまで 5 分待機します。合計実行時間: 11 分。これは 15 分のデフォルト制限を下回っているため、両方のキャパシティモードで機能します。

  • Lambda Managed Instances のみで有効: 耐久関数は Amazon SQS メッセージを処理し、最初の処理を 2 分で完了してから、外部コールバックが完了するまで 20 分待機します。合計実行時間: 22 分。これは 15 分のデフォルト制限を超えていますが、Lambda マネージドインスタンスで実行されている関数の 90 分の制限内に収まっています。

  • 無効 (任意のキャパシティモード): 耐久関数は、手順間に合計 2 時間におよぶ複数の待機操作を伴いつつ、Kinesis レコードを処理します。これは 90 分の Lambda マネージドインスタンスの制限も超えるため、キャパシティモードに関係なく実行は失敗します。このような長いワークフローには、中間関数パターンを使用してください。

重要

イベントソースマッピングを使用する場合は、適用される制限内で耐久性のある実行のタイムアウトを設定してください。設定しない場合、イベントソースマッピングの作成に失敗します。デフォルトの制限時間は 15 分ですが、Lambda マネージドインスタンスでは最大 90 分です。ワークフローで実行時間が長い場合は、以下で説明する中間関数パターンを使用します。

イベントソースマッピングの設定

Lambda コンソール、AWS CLI、または AWS SDK を使用して、耐久性のある関数のイベントソースマッピングを設定します。すべての標準イベントソースマッピングのプロパティは、耐久関数に適用されます。

aws lambda create-event-source-mapping \ --function-name arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1 \ --event-source-arn arn:aws:sqs:us-east-1:123456789012:my-queue \ --batch-size 10 \ --maximum-batching-window-in-seconds 5

耐久関数のイベントソースマッピングを設定するときは、必ず修飾 ARN (バージョン番号またはエイリアス) を使用してください。

イベントソースマッピングによるエラー処理

イベントソースマッピングは、耐久関数で動作する組み込みのエラー処理を提供します。

  • 再試行動作: 最初の呼び出しが失敗した場合、イベントソースマッピングは再試行設定に従って再試行します。要件に基づいて最大再試行回数と再試行間隔を設定します。

  • デッドレターキュー: すべての再試行後に失敗したレコードをキャプチャするようにデッドレター キューを構成します。これにより、メッセージの損失を防ぎ、失敗したレコードを手動で検査できます。

  • 部分的なバッチ処理の失敗: Amazon SQS および Kinesis では、部分的なバッチ処理の失敗レポートを使用して、レコードを個別に処理し、失敗したレコードのみを再試行します。

  • Bisect on error: Kinesis および DynamoDB Streams の場合、bisect on error 機能を有効にして、失敗したバッチを分割し、問題のあるレコードを分離します。

注記

耐久性のある関数では、エラー処理用のデッドレターキュー (DLQ) がサポートされていますが、Lambda 送信先はサポートされていません。DLQ を設定し、失敗した呼び出しのレコードをキャプチャします。

イベントソースマッピングエラー処理の詳細については、「イベントソースマッピング」を参照してください。

長時間実行されるワークフローに中間関数を使用する

ワークフローが該当するイベントソースマッピングの制限 (デフォルトでは 15 分、Lambda マネージドインスタンスでは 90 分) を超える場合は、イベントソースマッピングと耐久性のある関数の間に中間の標準 Lambda 関数を使用してください。中間関数は、イベントソースマッピングからイベントを受信し、耐久関数を非同期的に呼び出すことで、実行期間の制限を取り除きます。

このパターンは、イベントソースマッピングの同期呼び出しモデルを耐久関数の長時間かかる実行モデルから切り離します。イベントソースマッピングは中間関数を呼び出し、中間関数は耐久性のある実行を開始した後すぐに返されます。その後、耐久関数は、必要な期間 (最大 1 年間) 独立して実行されます。

アーキテクチャ

中間関数パターンでは、次の 3 つのコンポーネントを使用します。

  1. イベントソースマッピング: イベントソース (Amazon SQS、Kinesis、DynamoDB Streams) をポーリングし、レコードのバッチと同期的に中間関数を呼び出します。

  2. 中間関数: イベントソースマッピングからイベントを受信し、必要に応じてデータを検証および変換し、耐久関数を非同期的に呼び出す標準の Lambda 関数。この関数は迅速に (通常は 1 秒未満) 完了し、イベントソースマッピングに制御を返します。

  3. 耐久関数: 長時間の実行が可能な複雑かつ多段階のロジックでイベントを処理します。非同期的に呼び出されるため、イベントソースマッピングの実行時間制限による制約を受けません。

実装

中間関数は、イベントソースマッピングからイベント全体を受信し、耐久関数を非同期的に呼び出します。実行名パラメータを使用してべき等実行が開始されるようにし、イベントソースマッピングが再試行された場合に重複処理を防止します。

TypeScript
import { LambdaClient, InvokeCommand } from '@aws-sdk/client-lambda'; import { SQSEvent } from 'aws-lambda'; import { createHash } from 'crypto'; const lambda = new LambdaClient({}); export const handler = async (event: SQSEvent) => { // Invoke durable function asynchronously with execution name await lambda.send(new InvokeCommand({ FunctionName: 'arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1', InvocationType: 'Event', Payload: JSON.stringify({ executionName: event.Name, event: event }) })); return { statusCode: 200 }; };
Python
import boto3 import json import hashlib lambda_client = boto3.client('lambda') def handler(event, context): # Invoke durable function asynchronously with execution name lambda_client.invoke( FunctionName='arn:aws:lambda:us-east-1:123456789012:function:my-durable-function:1', InvocationType='Event', Payload=json.dumps({ 'executionName': execution_name, 'event': event["name"] }) ) return {'statusCode': 200}

中間関数自体のべき等性については、Powertools for AWS Lambda を使用して、イベントソースマッピングが中間関数を再試行した場合に、耐久関数の重複呼び出しを防止します。

耐久関数は、実行名を持つペイロードを受信し、長時間実行されるロジックを持つすべてのレコードを処理します。

TypeScript
import { withDurableExecution, DurableContext } from '@aws/durable-execution-sdk-js'; export const handler = withDurableExecution( async (payload: any, context: DurableContext) => { const sqsEvent = payload.event; // Process each record with complex, multi-step logic const results = await context.map( sqsEvent.Records, async (ctx, record) => { const validated = await ctx.step('validate', async () => { return validateOrder(JSON.parse(record.body)); }); // Wait for external approval (could take hours or days) const approval = await ctx.waitForCallback( 'approval', async (callbackId) => { await requestApproval(callbackId, validated); }, { timeout: { hours: 48 } } ); // Complete processing return await ctx.step('complete', async () => { return completeOrder(validated, approval); }); } ); return { statusCode: 200, processed: results.getResults().length }; } );
Python
from aws_durable_execution_sdk_python import durable_execution, DurableContext from aws_durable_execution_sdk_python.config import Duration, WaitForCallbackConfig from collections.abc import Sequence import json def validate_order(order_data: dict) -> dict: """Validate order data - always passes.""" return order_data def request_approval(callback_id: str, validated_order: dict) -> None: """Request approval for the order - always passes.""" pass def complete_order(validated_order: dict, approval_result: str) -> dict: """Complete the order processing - always passes.""" return validated_order @durable_execution def lambda_handler(payload, context: DurableContext): sqs_event = payload['event'] def process_record( ctx: DurableContext, record: dict, index: int, items: Sequence[dict] ) -> dict: validated = ctx.step( lambda _: validate_order(json.loads(record['body'])), name=f'validate-{index}' ) approval = ctx.wait_for_callback( submitter=lambda callback_id, wait_ctx: request_approval(callback_id, validated), name=f'approval-{index}', config=WaitForCallbackConfig(timeout=Duration.from_seconds(172800)) ) return ctx.step( lambda _: complete_order(validated, approval), name=f'complete-{index}' ) results = context.map( inputs=sqs_event['Records'], func=process_record, name='process-records' ) return { 'statusCode': 200, 'started': results.started_count, 'completed': results.success_count, 'failed': results.failure_count, 'total': results.total_count }

主な考慮事項

このパターンは、イベントソースマッピングと永続的な実行を切り離すことで、イベントソースマッピングの実行時間制限を解消します。中間関数は、耐久性のある実行を開始した直後に返り、イベントソースマッピングの処理を続行できるようにします。その後、耐久関数は、必要な期間だけ独立して実行されます。

中間関数は、耐久性のある実行が完了したときではなく、耐久関数を呼び出すときに成功します。耐久性のある実行が後で失敗した場合、イベントソースマッピングはすでにバッチを正常に処理しているため、再試行されません。耐久関数にエラー処理を実装し、失敗した実行に対してデッドレターキューを設定します。

実行名パラメータを使用して、べき等実行が開始されるようにします。イベントソースマッピングが中間関数を再試行する場合、実行名がすでに存在するため、耐久関数は重複した実行を開始しません。

サポートされているイベントソース

耐久関数は、イベントソースマッピングを使用するすべての Lambda イベントソースをサポートします。

  • Amazon SQS キュー (スタンダードおよび FIFO)

  • Kinesis Streams

  • DynamoDB Streams

  • Amazon Managed Streaming for Apache Kafka (Amazon MSK)

  • セルフマネージド Apache Kafka

  • Amazon MQ (ActiveMQ および RabbitMQ)

  • Amazon DocumentDB 変更ストリーム

耐久関数を呼び出す場合、すべてのイベントソースタイプには、前述の耐久性のある実行の制限が適用されます。Lambda マネージドインスタンスでは、Amazon MQ と Amazon DocumentDB を除き、これらのイベントソースの制限は 90 分に増加します。Amazon MQ と Amazon DocumentDB は引き続き 15 分に制限されます。