Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Mappature delle sorgenti di eventi con funzioni durevoli

Le funzioni durevoli funzionano con tutte le mappature delle sorgenti di eventi Lambda. Configura le mappature delle sorgenti di eventi per le funzioni durevoli nello stesso modo in cui le configuri per le funzioni standard. Le mappature delle sorgenti di eventi eseguono automaticamente il polling di sorgenti di eventi come Amazon SQS, Kinesis e DynamoDB Streams e richiamano la tua funzione con batch di record.

Le mappature delle sorgenti di eventi sono utili per funzioni durevoli che elaborano flussi o code con flussi di lavoro complessi in più fasi. Ad esempio, puoi creare una funzione duratura che elabori i messaggi Amazon SQS con tentativi, chiamate API esterne e approvazioni umane.

In che modo le mappature delle sorgenti degli eventi richiamano funzioni durevoli

Le mappature delle sorgenti di eventi richiamano le funzioni durevoli in modo sincrono, aspettando il completamento dell'esecuzione durevole completa prima di elaborare il batch successivo o contrassegnare i record come elaborati. Se il tempo totale di esecuzione durevole supera i 15 minuti, l'esecuzione scade e non riesce. La mappatura dell'origine dell'evento riceve un'eccezione di timeout e la gestisce in base alla configurazione del nuovo tentativo.

Limite di esecuzione di 15 minuti

Quando le funzioni durevoli vengono richiamate dalle mappature delle sorgenti degli eventi, la durata totale dell'esecuzione durevole non può superare i 15 minuti. Questo limite si applica all'intera esecuzione duratura dall'inizio al completamento, non solo alle singole chiamate di funzioni.

Questo limite di 15 minuti è separato dal timeout della funzione Lambda (anch'esso massimo di 15 minuti). Il timeout della funzione controlla la durata di ogni singola chiamata, mentre il timeout di esecuzione durevole controlla il tempo totale trascorso dall'inizio al completamento dell'esecuzione.

Scenari di esempio:

Configurazione delle mappature delle sorgenti degli eventi

Configura le mappature delle sorgenti di eventi per funzioni durature utilizzando la console Lambda o gli SDK. AWS CLI AWS Tutte le proprietà standard di mappatura delle sorgenti di eventi si applicano alle funzioni durevoli:

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
    

Ricorda di utilizzare un ARN qualificato (con numero di versione o alias) quando configuri le mappature delle sorgenti di eventi per funzioni durevoli.

Gestione degli errori con le mappature delle sorgenti degli eventi

Le mappature delle sorgenti degli eventi forniscono una gestione degli errori integrata che funziona con funzioni durevoli:

Per informazioni complete sulla gestione degli errori di mappatura delle sorgenti degli eventi, vedere mappature delle sorgenti degli eventi.

Utilizzo di una funzione intermedia per flussi di lavoro di lunga durata

Se il completamento del flusso di lavoro richiede più di 15 minuti, utilizza una funzione Lambda standard intermedia tra la mappatura dell'origine dell'evento e la funzione durevole. La funzione intermediaria riceve gli eventi dalla mappatura dell'origine degli eventi e richiama la funzione durable in modo asincrono, rimuovendo il limite di esecuzione di 15 minuti.

Questo modello separa il modello di invocazione sincrona della mappatura dell'origine degli eventi dal modello di esecuzione a lungo termine della funzione durable. La mappatura dell'origine degli eventi richiama la funzione intermedia, che ritorna rapidamente dopo l'avvio dell'esecuzione durevole. La funzione durable viene quindi eseguita in modo indipendente per tutto il tempo necessario (fino a 1 anno).

Architecture

Il modello di funzione intermedio utilizza tre componenti:

Implementazione

La funzione intermediaria riceve l'intero evento dalla mappatura della sorgente dell'evento e richiama la funzione durable in modo asincrono. Utilizzate il parametro execution name per assicurarvi che l'esecuzione venga avviata in modo idempotente, impedendo l'elaborazione duplicata se la mappatura della sorgente dell'evento riprova:

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}
            

Per verificare l'idempotenza della funzione intermediaria stessa, utilizzate Powertools per evitare invocazioni duplicate della funzione durable se la AWS Lambda mappatura della sorgente dell'evento riprova la funzione intermediaria.

La funzione durable riceve il payload con il nome dell'esecuzione ed elabora tutti i record con una logica a lungo termine:

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
    }
            

Considerazioni chiave

Questo modello rimuove il limite di esecuzione di 15 minuti disaccoppiando la mappatura della sorgente dell'evento dall'esecuzione durevole. La funzione intermediaria ritorna immediatamente dopo l'avvio dell'esecuzione durevole, consentendo alla mappatura della sorgente dell'evento di continuare l'elaborazione. La funzione durable viene quindi eseguita in modo indipendente per tutto il tempo necessario.

La funzione intermediaria ha successo quando richiama la funzione durevole, non quando l'esecuzione durevole viene completata. Se l'esecuzione durevole fallisce in un secondo momento, la mappatura dell'origine dell'evento non verrà riprovata perché il batch è già stato elaborato correttamente. Implementa la gestione degli errori nella funzione durable e configura le code di lettere morte per le esecuzioni non riuscite.

Utilizzate il parametro execution name per assicurarvi che l'esecuzione abbia inizio in modo idempotente. Se la mappatura dell'origine dell'evento riprova la funzione intermediaria, la funzione durable non avvierà un'esecuzione duplicata perché il nome dell'esecuzione esiste già.

Fonti di eventi supportate

Le funzioni Durable supportano tutte le sorgenti di eventi Lambda che utilizzano le mappature delle sorgenti di eventi:

Tutti i tipi di sorgenti di eventi sono soggetti al limite di esecuzione duratura di 15 minuti quando si richiamano funzioni durevoli.