AWS Lambda 함수 호출

AWS Lambda 함수 호출 (Invoke AWS Lambda function)

배치 작업 중에 객체마다 커스텀 처리를 하고 싶을 때가 있죠. Invoke AWS Lambda function 작업은 매니페스트에 있는 객체마다 Lambda 함수를 실행해서, S3만으로는 하기 어려운 맞춤 로직을 적용할 수 있게 해 줍니다.

출처: 문서

본문

Amazon S3 Batch Operations를 쓰면 S3 객체에 대해 대규모 배치 작업을 수행할 수 있어요. Invoke AWS Lambda function Batch Operations 작업은 매니페스트에 나열된 객체에서 커스텀 작업을 수행하기 위해 AWS Lambda 함수를 시작해요. 이 절은 S3 Batch Operations와 함께 사용할 Lambda 함수를 만드는 방법과, 그 함수를 호출하는 작업을 만드는 방법을 설명해요. S3 Batch Operations 작업은 LambdaInvoke 작업을 사용해 매니페스트에 나열된 모든 객체에 대해 Lambda 함수를 실행해요.

S3 Batch Operations는 Amazon S3 콘솔, AWS Command Line Interface(AWS CLI), AWS SDK, Amazon S3 REST API로 사용할 수 있어요. Lambda 사용에 대한 자세한 내용은 AWS Lambda Developer Guide의 Getting Started with AWS Lambda 문서를 참고하세요.

아래 절에서 S3 Batch Operations를 Lambda와 함께 시작하는 방법을 설명할게요.

Lambda를 Batch Operations와 함께 사용하기
S3 Batch Operations에 사용할 Lambda 함수 만들기
Lambda 함수를 호출하는 S3 Batch Operations 작업 만들기
Lambda 매니페스트에서 태스크 수준 정보 제공
S3 Batch Operations 튜토리얼

Lambda를 Batch Operations와 함께 사용하기

S3 Batch Operations와 AWS Lambda를 사용할 때는 S3 Batch Operations 전용으로 새 Lambda 함수를 만들어야 해요. 기존 Amazon S3 이벤트 기반 함수를 S3 Batch Operations에 재사용할 수 없어요. 이벤트 함수는 메시지를 받기만 하고 메시지를 반환하지 않으니까요. S3 Batch Operations에 사용하는 Lambda 함수는 메시지를 받고 반환해야 해요. Amazon S3 이벤트와 함께 Lambda를 사용하는 방법은 AWS Lambda Developer Guide의 Using AWS Lambda with Amazon S3 문서를 참고하세요.

Lambda 함수를 호출하는 S3 Batch Operations 작업을 만들어요. 작업은 매니페스트에 나열된 모든 객체에서 같은 Lambda 함수를 실행해요. 매니페스트의 객체를 처리하는 동안 Lambda 함수의 어떤 버전을 사용할지 제어할 수 있어요. S3 Batch Operations는 비정규화된(qualified) Amazon Resource Name(ARN), 별칭(alias), 특정 버전을 지원해요. 자세한 내용은 AWS Lambda Developer Guide의 Introduction to AWS Lambda Versioning 문서를 참고하세요.

S3 Batch Operations 작업에 별칭이나 $LATEST 한정자를 사용하는 함수 ARN을 제공했는데 둘 중 하나가 가리키는 버전을 갱신하면, S3 Batch Operations는 새 버전의 Lambda 함수를 호출하기 시작해요. 큰 작업을 진행하는 중간에 기능을 갱신하고 싶을 때 유용하죠. S3 Batch Operations가 사용하는 버전을 바꾸고 싶지 않다면, 작업을 만들 때 FunctionARN 매개변수에 특정 버전을 제공하면 돼요.

S3 Batch Operations와 함께하는 단일 AWS Lambda 작업은 최대 200억 개 객체의 매니페스트를 지원할 수 있어요.

디렉터리 버킷에서 Lambda와 Batch Operations 사용하기

디렉터리 버킷(directory bucket)은 일관된 한 자리 숫자 밀리초 대기 시간이 필요한 워크로드나 성능 중심 애플리케이션을 위해 설계된 Amazon S3 버킷 유형이에요. 자세한 내용은 Directory buckets 문서를 참고하세요.

Batch Operations로 디렉터리 버킷에 작용하는 Lambda 함수를 호출할 때는 특별한 요구 사항이 있어요. 예를 들어 갱신된 JSON 스키마를 사용해 Lambda 요청을 구성하고, 작업을 만들 때 InvocationSchemaVersion 2.0(1.0이 아니라)을 지정해야 해요. 이 갱신된 스키마에서는 UserArguments에 선택적 키-값 쌍을 지정할 수 있는데, 이를 사용해 기존 Lambda 함수의 특정 매개변수를 수정할 수 있어요. 자세한 내용은 AWS Storage Blog의 Automate object processing in Amazon S3 directory buckets with S3 Batch Operations and AWS Lambda 문서를 참고하세요.

응답 및 결과 코드 (Response and result codes)

S3 Batch Operations는 하나 이상의 키로 Lambda 함수를 호출하며, 각 키에는 TaskID가 연결돼요. S3 Batch Operations는 Lambda 함수에서 키별 결과 코드를 기대해요. 요청에서 보낸 작업 ID 중 per-key 결과 코드와 함께 반환되지 않은 것에는 treatMissingKeysAs 필드의 결과 코드가 부여돼요. treatMissingKeysAs는 선택적 요청 필드이며 기본값은 TemporaryFailure예요. 다음 표는 treatMissingKeysAs 필드의 다른 가능한 결과 코드와 값을 보여줘요.

응답 코드	설명
Succeeded	태스크가 정상적으로 완료됐어요. 작업 완료 보고서를 요청했다면 태스크의 결과 문자열이 보고서에 포함돼요.
TemporaryFailure	태스크가 일시적 실패를 겪었고, 작업이 완료되기 전에 다시 실행(redrive)돼요. 결과 문자열은 무시돼요. 이게 마지막 재실행이라면 오류 메시지가 최종 보고서에 포함돼요.
PermanentFailure	태스크가 영구적 실패를 겪었어요. 작업 완료 보고서를 요청했다면 태스크는 `Failed`로 표시되고 오류 메시지 문자열이 포함돼요. 실패한 태스크의 결과 문자열은 무시돼요.

S3 Batch Operations에 사용할 Lambda 함수 만들기

이 절은 Lambda 함수와 함께 사용해야 하는 예제 AWS Identity and Access Management(IAM) 권한을 제공해요. 또 S3 Batch Operations와 함께 사용할 예제 Lambda 함수도 포함돼요. Lambda 함수를 한 번도 만들어 본 적이 없다면 AWS Lambda Developer Guide의 Tutorial: Using AWS Lambda with Amazon S3 문서를 참고하세요.

S3 Batch Operations 전용으로 Lambda 함수를 만들어야 해요. 기존 Amazon S3 이벤트 기반 Lambda 함수는 재사용할 수 없어요. S3 Batch Operations에 사용되는 Lambda 함수는 특수한 데이터 필드를 받고 반환해야 하기 때문이에요.

중요 Java로 작성된 AWS Lambda 함수는 RequestHandler 또는 RequestStreamHandler 핸들러 인터페이스를 받아들여요. 하지만 S3 Batch Operations 요청·응답 형식을 지원하려면 AWS Lambda는 요청과 응답의 커스텀 직렬화·역직렬화를 위해 RequestStreamHandler 인터페이스를 요구해요. 이 인터페이스는 Lambda가 Java handleRequest 메서드에 InputStream과 OutputStream을 전달할 수 있게 해 줘요. Lambda 함수를 S3 Batch Operations와 사용할 때는 반드시 RequestStreamHandler 인터페이스를 사용하세요. RequestHandler 인터페이스를 사용하면 완료 보고서에 "Invalid JSON returned in Lambda payload" 오류가 나며 배치 작업이 실패해요. 자세한 내용은 AWS Lambda User Guide의 Handler interfaces 문서를 참고하세요.

예제 IAM 권한

다음은 Lambda 함수를 S3 Batch Operations와 함께 사용하는 데 필요한 IAM 권한 예시예요.

다음은 Batch Operations IAM 역할에 사용할 수 있는 신뢰 정책 예시예요. 이 IAM 역할은 작업을 만들 때 지정하며, Batch Operations에게 IAM 역할을 맡을(assume) 권한을 부여해요.

{
    "Version":"2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "batchoperations.s3.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

다음은 S3 Batch Operations에게 Lambda 함수 호출과 입력 매니페스트 읽기 권한을 주는 IAM 정책 예시예요.

{
    "Version":"2012-10-17",
    "Statement": [
        {
            "Sid": "BatchOperationsLambdaPolicy",
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:GetObjectVersion",
                "s3:PutObject",
                "lambda:InvokeFunction"
            ],
            "Resource": "*"
        }
    ]
}

예제 요청 및 응답

이 절은 Lambda 함수에 대한 요청과 응답 예제를 제공해요.

다음은 Lambda 함수에 대한 요청의 JSON 예시예요.

{
    "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"
        }
    ]
}

다음은 Lambda 함수에 대한 응답의 JSON 예시예요.

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

S3 Batch Operations용 예제 Lambda 함수

다음 예제 Python Lambda는 버전 관리되는 객체에서 삭제 마커(delete marker)를 제거해요.

예제에서 보듯, S3 Batch Operations의 키는 URL 인코딩되어 있어요. Amazon S3를 다른 AWS 서비스와 함께 사용하려면 S3 Batch Operations에서 전달되는 키를 URL 디코딩하는 것이 중요해요.

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,
    }

Lambda 함수를 호출하는 S3 Batch Operations 작업 만들기

Lambda 함수를 호출하는 S3 Batch Operations 작업을 만들 때는 다음을 제공해야 해요.

Lambda 함수의 ARN(함수 별칭 또는 특정 버전 번호를 포함할 수 있어요)
함수를 호출할 권한이 있는 IAM 역할
LambdaInvokeFunction 작업 매개변수

S3 Batch Operations 작업 만들기에 대한 자세한 내용은 Creating an S3 Batch Operations job 및 Operations supported by S3 Batch Operations 문서를 참고하세요.

다음 예제는 AWS CLI로 Lambda 함수를 호출하는 S3 Batch Operations 작업을 만들어요. 이 예제를 사용하려면 사용자 입력 플레이스홀더를 자신의 정보로 바꾸세요.

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"

Lambda 매니페스트에서 태스크 수준 정보 제공하기

AWS Lambda 함수를 S3 Batch Operations와 함께 사용할 때, 작용하는 각 태스크나 키에 추가 데이터를 함께 전달하고 싶을 수 있어요. 예를 들어 소스 객체 키와 새 객체 키를 모두 제공하고 싶을 수 있죠. 그러면 Lambda 함수가 소스 키를 새 이름으로 다른 S3 버킷에 복사할 수 있어요. 기본적으로 Batch Operations는 작업에 대한 입력 매니페스트에서 대상 버킷과 소스 키 목록만 지정할 수 있게 해 줘요. 다음 예제들은 더 복잡한 Lambda 함수를 실행할 수 있도록 매니페스트에 추가 데이터를 포함하는 방법을 설명해요.

Lambda 함수 코드에서 사용할 S3 Batch Operations 매니페스트에 키별 매개변수를 지정하려면 다음 URL 인코딩된 JSON 형식을 사용해요. key 필드는 Amazon S3 객체 키인 것처럼 Lambda 함수에 전달돼요. 하지만 아래 예제처럼 Lambda 함수가 이 필드를 다른 값이나 여러 키를 담도록 해석할 수 있어요.

참고 매니페스트의 key 필드에 사용할 수 있는 최대 문자 수는 1,024자예요.

URL 인코딩된 버전을 S3 Batch Operations에 제공해야 해요.

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

이 URL 인코딩된 버전을 S3 Batch Operations에 제공해야 해요. URL 인코딩되지 않은 버전은 작동하지 않아요.

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

이 URL 인코딩된 매니페스트 예제는 다음 Lambda 함수가 파싱하도록 파이프로 구분된 객체 키를 포함해요.

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

이 Lambda 함수는 S3 Batch Operations 매니페스트에 인코딩된 파이프로 구분된 태스크를 파싱하는 방법을 보여줘요. 태스크는 지정된 객체에 어떤 개정(revision) 작업이 적용되는지 나타내요.

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,
    }

S3 Batch Operations 튜토리얼

다음 튜토리얼은 Lambda로 일부 Batch Operations 태스크를 수행하는 전체 엔드투엔드 절차를 제시해요. 이 튜토리얼에서는 S3 소스 버킷에 저장된 비디오를 일괄 트랜스코딩하기 위해 Batch Operations가 Lambda 함수를 호출하도록 설정하는 방법을 배워요. Lambda 함수는 AWS Elemental MediaConvert를 호출해 비디오를 트랜스코딩해요.

튜토리얼: S3 Batch Operations로 비디오 일괄 트랜스코딩