Appel de la fonction AWS Lambda

Vous pouvez utiliser Amazon S3 Batch Operations pour effectuer des opérations par lot à grande échelle au niveau d’objets Amazon S3. L’opération Invoquer la fonction AWS Lambda de Batch Operations lance des fonctions AWS Lambda afin d’effectuer des actions personnalisées sur des objets répertoriés dans un manifeste. Cette section décrit comment créer une fonction Lambda à utiliser avec S3 Batch Operations et comment créer une tâche pour invoquer la fonction. La tâche S3 Batch Operations utilise l’opération LambdaInvoke pour exécuter une fonction Lambda sur chaque objet répertorié dans un manifeste.

Vous pouvez créer des tâches S3 Batch Operations via la console Amazon S3, l’AWS Command Line Interface (AWS CLI), les kits AWS SDK ou l’API REST Amazon S3. Pour plus d’informations sur l’utilisation de Lambda, consultez Mise en route avec AWS Lambda dans le Guide du développeur AWS Lambda.

Les sections suivantes expliquent comment commencer à utiliser S3 Batch Operations avec Lambda.

Utilisation de Lambda avec Batch Operations

Lorsque vous utilisez S3 Batch Operations avec AWS Lambda, vous devez créer de nouvelles fonctions Lambda spécifiquement destinées à être utilisées avec S3 Batch Operations. Vous ne pouvez pas réutiliser les fonctions basées sur des événements Amazon S3 existantes avec S3 Batch Operations. Les fonctions d’événements peuvent seulement recevoir des messages. Elles ne peuvent pas en renvoyer. Les fonctions Lambda utilisées avec S3 Batch Operations doivent accepter et renvoyer des messages. Pour plus d’informations sur l’utilisation de Lambda avec des événements Amazon S3, consultez Utilisation d’AWS Lambda avec Amazon S3 dans le Guide du développeur AWS Lambda.

Vous créez une tâche S3 Batch Operations qui invoque votre fonction Lambda. La tâche exécute la même fonction Lambda pour tous les objets répertoriés dans votre manifeste. Vous pouvez contrôler les versions de votre fonction Lambda à utiliser lors du traitement des objets dans votre manifeste. S3 Batch Operations prend en charge les Amazon Resource Names (ARN) non qualifiés, les alias et les versions spécifiques. Pour plus d’informations, consultez Présentation de la AWS Lambdagestion des versions dans le Guide du développeur AWS Lambda.

Si vous donnez un ARN de fonction qui utilise un alias ou le qualificatif $LATEST à la tâche S3 Batch Operations, et que vous mettez à jour la version vers laquelle l’un de ces points pointe, S3 Batch Operations commence à appeler la nouvelle version de votre fonction Lambda. Cela peut être utile lorsque vous souhaitez mettre à jour la fonctionnalité en cours de traitement d’une tâche importante. Si vous ne souhaitez pas que S3 Batch Operations modifie la version utilisée, indiquez la version spécifique dans le paramètre FunctionARN lorsque vous créez la tâche.

Une seule tâche AWS Lambda avec S3 Batch Operations peut prendre en charge un manifeste contenant jusqu’à 20 milliards d’objets.

Utilisation de Lambda et de Batch Operations avec des compartiments de répertoires

Les compartiments de répertoires sont un type de compartiment Amazon S3 conçu pour les charges de travail ou les applications critiques en termes de performances qui nécessitent une latence constante inférieure à dix millisecondes. Pour plus d’informations, consultez Compartiments de répertoires.

L’utilisation de Batch Operations pour invoquer des fonctions Lambda agissant sur des compartiments de répertoires est soumise à des exigences particulières. Par exemple, vous devez structurer votre demande Lambda à l’aide d’un schéma JSON mis à jour et spécifier InvocationSchemaVersion 2.0 (pas 1.0) lorsque vous créez la tâche. Ce schéma mis à jour vous permet de spécifier des paires clé-valeur facultatives pour UserArguments, que vous pouvez utiliser pour modifier certains paramètres des fonctions Lambda existantes. Pour plus d’informations, consultez Automatisation du traitement des objets dans les compartiments de répertoires Amazon S3 avec S3 Batch Operations et AWS Lambda dans le blog AWS Storage.

Codes de réponse et de résultat

S3 Batch Operations invoque la fonction Lambda avec une ou plusieurs clés, chacune étant associée à un TaskID. S3 Batch Operations s’attend à recevoir un code de résultat par clé de la part des fonctions Lambda. Tous les identifiants de tâche envoyés dans la demande qui ne sont pas renvoyés avec un code de résultat par clé recevront le code de résultat du champ treatMissingKeysAs. treatMissingKeysAs est un champ de demande facultatif dont la valeur par défaut correspond à TemporaryFailure. Le tableau suivant contient les autres codes de résultat et valeurs possibles pour le champ treatMissingKeysAs.

Création d’une fonction Lambda à utiliser avec S3 Batch Operations

Cette section fournit des exemples d’autorisations Gestion des identités et des accès AWS (IAM) que vous devez utiliser avec votre fonction Lambda. Elle contient également un exemple de fonction Lambda à utiliser avec S3 Batch Operations. Si vous n’avez jamais créé de fonction Lambda, consultez Didacticiel : Utilisation d’AWS Lambda avec Amazon S3 dans le Guide du développeur AWS Lambda.

Vous devez créer des fonctions Lambda spécifiques à utiliser avec S3 Batch Operations. Vous ne pouvez pas réutiliser les fonctions Lambda basées sur des événements Amazon S3, car les fonctions Lambda utilisées pour S3 Batch Operations doivent accepter et renvoyer des champs de données spéciaux.

Exemples d’autorisations IAM

Voici des exemples d’autorisations IAM nécessaires à l’utilisation d’une fonction Lambda avec S3 Batch Operations.

Exemple de demande et de réponse

Cette section fournit des exemples de demandes et réponses pour la fonction Lambda.

{
    "invocationSchemaVersion": "1.0",
    "invocationId": "YXNkbGZqYWRmaiBhc2RmdW9hZHNmZGpmaGFzbGtkaGZza2RmaAo",
    "job": {
        "id": "f3cc4f60-61f6-4a2b-8a21-d07600c373ce"
    },
    "tasks": [
        {
            "taskId": "dGFza2lkZ29lc2hlcmUK",
            "s3Key": "customerImage1.jpg",
            "s3VersionId": "1",
            "s3BucketArn": "arn:aws:s3:us-east-1:0123456788:amzn-s3-demo-bucket1"
        }
    ]
}

{
  "invocationSchemaVersion": "1.0",
  "treatMissingKeysAs" : "PermanentFailure",
  "invocationId" : "YXNkbGZqYWRmaiBhc2RmdW9hZHNmZGpmaGFzbGtkaGZza2RmaAo",
  "results": [
    {
      "taskId": "dGFza2lkZ29lc2hlcmUK",
      "resultCode": "Succeeded",
      "resultString": "[\"Mary Major", \"John Stiles\"]"
    }
  ]
}

Exemple de fonction Lambda pour S3 Batch Operations

L’exemple suivant Python Lambda supprime un marqueur de suppression d’un objet versionné.

Comme l’exemple le montre, les clés S3 Batch Operations sont encodées par URL. Pour utiliser Amazon S3 avec d’autres services AWS, il est important de décoder l’URL de la clé transmise par S3 Batch Operations.

import logging
from urllib import parse
import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)
logger.setLevel("INFO")

s3 = boto3.client("s3")


def lambda_handler(event, context):
    """
    Removes a delete marker from the specified versioned object.

    :param event: The S3 batch event that contains the ID of the delete marker
                  to remove.
    :param context: Context about the event.
    :return: A result structure that Amazon S3 uses to interpret the result of the
             operation. When the result code is TemporaryFailure, S3 retries the
             operation.
    """
    # Parse job parameters from Amazon S3 batch operations
    invocation_id = event["invocationId"]
    invocation_schema_version = event["invocationSchemaVersion"]

    results = []
    result_code = None
    result_string = None

    task = event["tasks"][0]
    task_id = task["taskId"]

    try:
        obj_key = parse.unquote_plus(task["s3Key"], encoding="utf-8")
        obj_version_id = task["s3VersionId"]
        bucket_name = task["s3BucketArn"].split(":")[-1]

        logger.info(
            "Got task: remove delete marker %s from object %s.", obj_version_id, obj_key
        )

        try:
            # If this call does not raise an error, the object version is not a delete
            # marker and should not be deleted.
            response = s3.head_object(
                Bucket=bucket_name, Key=obj_key, VersionId=obj_version_id
            )
            result_code = "PermanentFailure"
            result_string = (
                f"Object {obj_key}, ID {obj_version_id} is not " f"a delete marker."
            )

            logger.debug(response)
            logger.warning(result_string)
        except ClientError as error:
            delete_marker = error.response["ResponseMetadata"]["HTTPHeaders"].get(
                "x-amz-delete-marker", "false"
            )
            if delete_marker == "true":
                logger.info(
                    "Object %s, version %s is a delete marker.", obj_key, obj_version_id
                )
                try:
                    s3.delete_object(
                        Bucket=bucket_name, Key=obj_key, VersionId=obj_version_id
                    )
                    result_code = "Succeeded"
                    result_string = (
                        f"Successfully removed delete marker "
                        f"{obj_version_id} from object {obj_key}."
                    )
                    logger.info(result_string)
                except ClientError as error:
                    # Mark request timeout as a temporary failure so it will be retried.
                    if error.response["Error"]["Code"] == "RequestTimeout":
                        result_code = "TemporaryFailure"
                        result_string = (
                            f"Attempt to remove delete marker from  "
                            f"object {obj_key} timed out."
                        )
                        logger.info(result_string)
                    else:
                        raise
            else:
                raise ValueError(
                    f"The x-amz-delete-marker header is either not "
                    f"present or is not 'true'."
                )
    except Exception as error:
        # Mark all other exceptions as permanent failures.
        result_code = "PermanentFailure"
        result_string = str(error)
        logger.exception(error)
    finally:
        results.append(
            {
                "taskId": task_id,
                "resultCode": result_code,
                "resultString": result_string,
            }
        )
    return {
        "invocationSchemaVersion": invocation_schema_version,
        "treatMissingKeysAs": "PermanentFailure",
        "invocationId": invocation_id,
        "results": results,
    }



                

Création d’une tâche S3 Batch Operations qui invoque une fonction Lambda

Lors de la création d’une tâche S3 Batch Operations pour appeler une fonction Lambda, vous devez fournir les éléments suivants :

Pour plus d’informations sur la création d’une tâche S3 Batch Operations, consultez Création d’une tâche S3 Batch Operations et Opérations prises en charge par S3 Batch Operations.

L’exemple suivant crée une tâche S3 Batch Operations qui invoque une fonction Lambda à l’aide de l’AWS CLI. Pour utiliser cet exemple, remplacez les user input placeholders par vos propres informations.

aws s3control create-job
    --account-id account-id
    --operation  '{"LambdaInvoke": { "FunctionArn": "arn:aws:lambda:region:account-id:function:LambdaFunctionName" } }'
    --manifest '{"Spec":{"Format":"S3BatchOperations_CSV_20180820","Fields":["Bucket","Key"]},"Location":{"ObjectArn":"arn:aws:s3:::amzn-s3-demo-manifest-bucket","ETag":"ManifestETag"}}'
    --report '{"Bucket":"arn:aws:s3:::amzn-s3-demo-bucket","Format":"Report_CSV_20180820","Enabled":true,"Prefix":"ReportPrefix","ReportScope":"AllTasks"}'
    --priority 2
    --role-arn arn:aws:iam::account-id:role/BatchOperationsRole
    --region region
    --description "Lambda Function"

Fourniture d’informations au niveau des tâches dans les manifestes Lambda

Lorsque vous utilisez les fonctions AWS Lambda avec S3 Batch Operations, vous pouvez souhaiter que des données supplémentaires accompagnent chaque tâche ou clé exploitée. Par exemple, vous pouvez souhaiter que la clé d’objet source et la nouvelle clé d’objet soient fournies. Votre fonction Lambda peut alors copier la clé source vers un nouveau compartiment S3 avec un nouveau nom. Par défaut, Batch Operations vous permet de spécifier uniquement le compartiment de destination et une liste de clés source dans le manifeste d’entrée dans votre tâche. Les exemples suivants décrivent comment inclure des données supplémentaires dans le manifeste afin que vous puissiez exécuter des fonctions Lambda plus complexes.

Pour spécifier des paramètres individuels pour chaque clé dans votre manifeste S3 Batch Operations à utiliser dans le code de votre fonction Lambda, utilisez le format JSON codé en URL suivant. Le champ key est transmis à votre fonction Lambda comme s’il s’agissait d’une clé d’objet Amazon S3. La fonction Lambda peut cependant considérer qu’il contient d’autres valeurs ou plusieurs clés, comme illustré dans les exemples suivants.

amzn-s3-demo-bucket,{"origKey": "object1key", "newKey": "newObject1Key"}
amzn-s3-demo-bucket,{"origKey": "object2key", "newKey": "newObject2Key"}
amzn-s3-demo-bucket,{"origKey": "object3key", "newKey": "newObject3Key"}

amzn-s3-demo-bucket,%7B%22origKey%22%3A%20%22object1key%22%2C%20%22newKey%22%3A%20%22newObject1Key%22%7D
amzn-s3-demo-bucket,%7B%22origKey%22%3A%20%22object2key%22%2C%20%22newKey%22%3A%20%22newObject2Key%22%7D
amzn-s3-demo-bucket,%7B%22origKey%22%3A%20%22object3key%22%2C%20%22newKey%22%3A%20%22newObject3Key%22%7D

amzn-s3-demo-bucket,object1key%7Clower
amzn-s3-demo-bucket,object2key%7Cupper
amzn-s3-demo-bucket,object3key%7Creverse
amzn-s3-demo-bucket,object4key%7Cdelete

import logging
from urllib import parse
import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)
logger.setLevel("INFO")

s3 = boto3.resource("s3")


def lambda_handler(event, context):
    """
    Applies the specified revision to the specified object.

    :param event: The Amazon S3 batch event that contains the ID of the object to
                  revise and the revision type to apply.
    :param context: Context about the event.
    :return: A result structure that Amazon S3 uses to interpret the result of the
             operation.
    """
    # Parse job parameters from Amazon S3 batch operations
    invocation_id = event["invocationId"]
    invocation_schema_version = event["invocationSchemaVersion"]

    results = []
    result_code = None
    result_string = None

    task = event["tasks"][0]
    task_id = task["taskId"]
    # The revision type is packed with the object key as a pipe-delimited string.
    obj_key, revision = parse.unquote_plus(task["s3Key"], encoding="utf-8").split("|")
    bucket_name = task["s3BucketArn"].split(":")[-1]

    logger.info("Got task: apply revision %s to %s.", revision, obj_key)

    try:
        stanza_obj = s3.Bucket(bucket_name).Object(obj_key)
        stanza = stanza_obj.get()["Body"].read().decode("utf-8")
        if revision == "lower":
            stanza = stanza.lower()
        elif revision == "upper":
            stanza = stanza.upper()
        elif revision == "reverse":
            stanza = stanza[::-1]
        elif revision == "delete":
            pass
        else:
            raise TypeError(f"Can't handle revision type '{revision}'.")

        if revision == "delete":
            stanza_obj.delete()
            result_string = f"Deleted stanza {stanza_obj.key}."
        else:
            stanza_obj.put(Body=bytes(stanza, "utf-8"))
            result_string = (
                f"Applied revision type '{revision}' to " f"stanza {stanza_obj.key}."
            )

        logger.info(result_string)
        result_code = "Succeeded"
    except ClientError as error:
        if error.response["Error"]["Code"] == "NoSuchKey":
            result_code = "Succeeded"
            result_string = (
                f"Stanza {obj_key} not found, assuming it was deleted "
                f"in an earlier revision."
            )
            logger.info(result_string)
        else:
            result_code = "PermanentFailure"
            result_string = (
                f"Got exception when applying revision type '{revision}' "
                f"to {obj_key}: {error}."
            )
            logger.exception(result_string)
    finally:
        results.append(
            {
                "taskId": task_id,
                "resultCode": result_code,
                "resultString": result_string,
            }
        )
    return {
        "invocationSchemaVersion": invocation_schema_version,
        "treatMissingKeysAs": "PermanentFailure",
        "invocationId": invocation_id,
        "results": results,
    }



 

Tutoriel de S3 Batch Operations

Le tutoriel suivant présente des procédures complètes de bout en bout pour certaines tâches Batch Operations avec Lambda. Dans ce didacticiel, vous apprendrez à configurer les opérations par lot S3 pour invoquer une fonction Lambda pour le transcodage par lot de vidéos stockées dans un compartiment source S3. La fonction Lambda appelle AWS Elemental MediaConvert pour transcoder les vidéos.