Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.

Pemetaan sumber acara dengan fungsi tahan lama

Fungsi tahan lama bekerja dengan semua pemetaan sumber peristiwa Lambda. Konfigurasikan pemetaan sumber peristiwa untuk fungsi tahan lama dengan cara yang sama seperti Anda mengonfigurasinya untuk fungsi standar. Pemetaan sumber peristiwa secara otomatis melakukan polling sumber acara seperti Amazon SQS, Kinesis, dan DynamoDB Streams, dan memanggil fungsi Anda dengan kumpulan catatan.

Pemetaan sumber peristiwa berguna untuk fungsi tahan lama yang memproses aliran atau antrian dengan alur kerja multi-langkah yang kompleks. Misalnya, Anda dapat membuat fungsi tahan lama yang memproses pesan Amazon SQS dengan percobaan ulang, panggilan API eksternal, dan persetujuan manusia.

Bagaimana pemetaan sumber peristiwa memanggil fungsi yang tahan lama

Pemetaan sumber peristiwa memanggil fungsi tahan lama secara serempak, menunggu eksekusi tahan lama lengkap selesai sebelum memproses batch berikutnya atau menandai catatan sebagai diproses. Jika total waktu eksekusi tahan lama melebihi 15 menit, waktu eksekusi habis dan gagal. Pemetaan sumber peristiwa menerima pengecualian batas waktu dan menanganinya sesuai dengan konfigurasi coba ulang.

Batas eksekusi 15 menit

Ketika fungsi tahan lama dipanggil oleh pemetaan sumber peristiwa, total durasi eksekusi tahan lama tidak boleh melebihi 15 menit. Batas ini berlaku untuk seluruh eksekusi tahan lama dari awal hingga selesai, bukan hanya pemanggilan fungsi individu.

Batas 15 menit ini terpisah dari batas waktu fungsi Lambda (juga maksimum 15 menit). Batas waktu fungsi mengontrol berapa lama setiap pemanggilan individu dapat berjalan, sedangkan batas waktu eksekusi yang tahan lama mengontrol total waktu yang telah berlalu dari awal eksekusi hingga penyelesaian.

Contoh skenario:

Mengkonfigurasi pemetaan sumber peristiwa

Konfigurasikan pemetaan sumber peristiwa untuk fungsi tahan lama menggunakan konsol Lambda, AWS CLI, atau SDK. AWS Semua properti pemetaan sumber peristiwa standar berlaku untuk fungsi tahan lama:

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
    

Ingatlah untuk menggunakan ARN yang memenuhi syarat (dengan nomor versi atau alias) saat mengonfigurasi pemetaan sumber peristiwa untuk fungsi tahan lama.

Penanganan kesalahan dengan pemetaan sumber peristiwa

Pemetaan sumber peristiwa menyediakan penanganan kesalahan bawaan yang berfungsi dengan fungsi tahan lama:

Untuk informasi lengkap tentang penanganan kesalahan pemetaan sumber peristiwa, lihat pemetaan sumber peristiwa.

Menggunakan fungsi perantara untuk alur kerja yang berjalan lama

Jika alur kerja Anda membutuhkan lebih dari 15 menit untuk diselesaikan, gunakan fungsi Lambda standar perantara antara pemetaan sumber peristiwa dan fungsi tahan lama Anda. Fungsi perantara menerima peristiwa dari pemetaan sumber peristiwa dan memanggil fungsi tahan lama secara asinkron, menghapus batas eksekusi 15 menit.

Pola ini memisahkan model pemanggilan sinkron pemetaan sumber peristiwa dari model eksekusi fungsi tahan lama yang berjalan. Pemetaan sumber peristiwa memanggil fungsi perantara, yang dengan cepat kembali setelah memulai eksekusi tahan lama. Fungsi tahan lama kemudian berjalan secara independen selama dibutuhkan (hingga 1 tahun).

Arsitektur

Pola fungsi perantara menggunakan tiga komponen:

Implementasi

Fungsi perantara menerima seluruh peristiwa dari pemetaan sumber peristiwa dan memanggil fungsi tahan lama secara asinkron. Gunakan parameter nama eksekusi untuk memastikan eksekusi idempotent dimulai, mencegah pemrosesan duplikat jika pemetaan sumber peristiwa mencoba lagi:

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}
            

Untuk idempotensi dalam fungsi perantara itu sendiri, gunakan Powertools for untuk AWS Lambda mencegah pemanggilan duplikat dari fungsi tahan lama jika pemetaan sumber peristiwa mencoba lagi fungsi perantara.

Fungsi tahan lama menerima payload dengan nama eksekusi dan memproses semua catatan dengan logika yang berjalan lama:

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
    }
            

Pertimbangan utama

Pola ini menghilangkan batas eksekusi 15 menit dengan memisahkan pemetaan sumber peristiwa dari eksekusi tahan lama. Fungsi perantara kembali segera setelah memulai eksekusi tahan lama, memungkinkan pemetaan sumber peristiwa untuk melanjutkan pemrosesan. Fungsi tahan lama kemudian berjalan secara independen selama diperlukan.

Fungsi perantara berhasil ketika memanggil fungsi tahan lama, bukan ketika eksekusi tahan lama selesai. Jika eksekusi tahan lama gagal nanti, pemetaan sumber peristiwa tidak akan dicoba lagi karena sudah berhasil memproses batch. Terapkan penanganan kesalahan dalam fungsi tahan lama dan konfigurasikan antrian surat mati untuk eksekusi yang gagal.

Gunakan parameter nama eksekusi untuk memastikan eksekusi idempotent dimulai. Jika pemetaan sumber peristiwa mencoba lagi fungsi perantara, fungsi tahan lama tidak akan memulai eksekusi duplikat karena nama eksekusi sudah ada.

Sumber acara yang didukung

Fungsi tahan lama mendukung semua sumber peristiwa Lambda yang menggunakan pemetaan sumber peristiwa:

Semua jenis sumber peristiwa tunduk pada batas eksekusi tahan lama 15 menit saat memanggil fungsi tahan lama.