Docs Menu
Docs Home
/
Atlas
/
Atlas Stream Processing
/
집계 파이프라인

$externalFunction

정의

$externalFunction

$externalFunction 단계는 특정 AWS Lambda 리소스에서 프로세스를 트리거합니다. AWS Lambda 프로세스에 대한 요청은 동기식 또는 비동기식일 수 있습니다.

AWS Lambda 생성 및 통합 AWS 액세스로 인증하기

Atlas Stream Processing 파이프라인 내에서 Amazon Web Services Lambda 리소스 호출하려면 Atlas Stream Processing 배포된 동일한 Amazon Web Services 리전 에 Amazon Web Amazon Web Services Lambda 배포해야 합니다. Amazon Web Services Lambda 리소스 배포에 대해 자세히 학습하려면 Amazon Web Services 설명서를 참조하세요.

1

Amazon Web Services Lambda 함수를 생성합니다.

Amazon Web Services CLI 또는 Amazon Web Services UI 사용하여 Lambda 함수를 생성합니다.

참고

  • AWS Lambda 리소스는 AWS_IAM AWS 인증 유형으로 생성해야 하며, 이를 위해 AWS IAM 역할 및 정책을 생성해야 합니다.

  • 버퍼링된 응답 유형만 지원되며, 스트리밍은 지원되지 않습니다.

2

통합 AWS 액세스를 구성하세요.

참고

여기에 설명된 절차는 Atlas UI의 기본 설정 흐름만을 다룹니다. 자세한 내용은 통합 AWS 액세스 설정 설명서를 참조하세요.

필요한 액세스 권한

통합 AWS 액세스를 설정하려면 프로젝트에 대한 Organization Owner 또는 Project Owner 액세스 권한이 있어야 합니다.

전제 조건

참고

  • AWS IAM 정책에 lambda:InvokeFunction 작업이 포함되어야 합니다.

  • 통합 AWS 액세스 구성 프로세스를 통해 사용할 수 있는 자리 표시자 ExternalId 및 Resource 값을 본인의 값으로 교체해야 합니다. 이 예시의 ExternalId에는 function-으로 시작하는 이름을 가진 모든 Lambda 함수와 일치하는 와일드카드가 포함되어 있음을 유의하세요.

3

Atlas UI에서 기존 역할에 신뢰 관계 추가하기

다음으로, AWS Lambda 리소스를 실행하려면 자체 관리형 AWS IAM 역할을 활성화해야 합니다.

permission-policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": "arn:aws:lambda:us-east-1:257394458927:function:<function-name>"
        }
    ]
}

  1. Atlas 프로젝트의 AWS IAM 통합 페이지로 이동하여 Authorize an AWS IAM role 버튼을 클릭합니다.

  2. 모달에 표시된 role-trust-policy.json 을 사용하여 새 역할을 생성하거나 기존 역할을 수정합니다.

  3. 역할이 생성되거나(또는 기존 역할이 새로운 신뢰 정책으로 업데이트되면) 해당 역할의 ARN을 모달에 붙여넣습니다.

  4. AWS 콘솔에서 IAM > Roles로 이동하여 역할을 선택합니다.

  5. permissions 탭에서 이 역할이 lambda를 호출할 수 있도록 새 '인라인 권한'을 추가합니다. 위의 permission-policy.json 예시에서는 <function-name> 이름의 모든 lambda를 실행할 수 있는 권한을 추가합니다.

  6. 마지막으로, Atlas Stream Processing 인스턴스로 이동하여 새 AWS Lambda connection을 추가하고 이전 단계에서 구성한 AWS IAM Role ARN을 선택합니다.

Atlas Stream Processing 인스턴스를 AWS Lambda 함수에 연결하기

Atlas Stream Processing 파이프라인 내에서 AWS Lambda 리소스로 요청을 보내려면 먼저 AWS Lambda 리소스를 Atlas Stream Processing 리소스의 연결로 추가해야 합니다.

다음 예시 와 같이 Atlas UI, Atlas CLI 또는 Atlas API 통해 Amazon Web Services Lambda 리소스 연결로 추가할 수 있습니다. Amazon Web Services IAM 구성의 arn (으)로 예시 의 roleArn 자리 표시자를 업데이트 할 수 있습니다.

curl --user "username:password" --digest \
   --header "Content-Type: application/json" \
   --header "Accept: application/vnd.atlas.2023-02-01+json" \
   --include \
   --data '{"name": "TestAWSLambdaConnection","type": "AWSLambda","aws": {"roleArn": "arn:aws:iam::<aws_account>:role/<role_name>"}}' \
   --request POST "https://cloud.mongodb.com/api/atlas/v2/groups/<group_id>/streams/<tenant_name>/connections"

구문

최소 요청

다음 예시에서는 최소 요청에 필요한 필드를 보여줍니다.

{ $externalFunction: {
  connectionName: "myLambdaConnection",
  functionName: "arn:aws:lambda:region:account-id:function:function-name",
  as: "response",
}}

맞춤형 요청

다음 사용자 정의 예제는 위에 설명된 필수 필드 외에도 오류 처리, 동기 실행 및 사전 처리된 Atlas Stream Processing 문서를 페이로드로 지정합니다.

{ $externalFunction: {
        connectionName: "myLambdaConnection",
        functionName: "arn:aws:lambda:region:account-id:function:function-name",
        execution: "sync"
        as: "response",
        onError: "fail",
        payload: [{$replaceRoot: { newRoot: "$fullDocument.payloadToSend" } }, { $addFields: { sum: { $sum: "$randomArray" }}}, { $project: { success: 1, sum: 1 }}],
}}

참고

onError 필드는 동기 및 비동기 요청 모두에 대한 AWS Lambda 리소스의 API 수준 오류 처리 방식과, 동기 요청에서 발생한 AWS Lambda 함수 오류 처리 방식을 정의합니다.

$externalFunction 단계에서는 다음 필드가 있는 문서를 사용합니다.

필드
유형
필요성
설명

connectionName

문자열

필수 사항

요청이 전송되는 연결 레지스트리에서 연결을 식별하는 레이블입니다.

functionName

문자열

필수 사항

전체 AWS ARN 또는 트리거할 AWS Lambda 함수의 이름입니다.

execution

열거형

옵션

Amazon Web Services Lambda 함수를 동기적으로 또는 비동기적으로 호출할지 여부를 지정하는 매개변수입니다. 허용되는 값은 다음과 같습니다.

  • sync

  • async

기본값은 sync이며, Atlas Stream Processing 파이프라인의 나머지 단계에서 AWS Lambda 함수의 출력이 필요한 경우 필수입니다.

as

문자열

옵션

REST API 응답의 필드 이름입니다.

엔드포인트가 0바이트를 반환하면, 연산자는 as 필드를 설정하지 않습니다.

onError

문자열

옵션

연산자에 HTTPS Status Code 또는 Lambda 런타임 관련 오류가 발생했을 때의 동작입니다. 다음 값 중 하나여야 합니다.

  • "dlq" : 영향을 받은 문서를 데드 레터 큐로 전달합니다.

  • "ignore" : 영향을 받는 문서에 대해 아무 작업도 하지 않습니다.

  • "fail" : 오류가 발생하면 스트림 프로세서를 종료합니다.

onError 는 $externalFunction 연산자 자체의 잘못된 구성으로 인해 발생하는 오류(예: 잘못된 표현식)에 대해서는 트리거되지 않습니다.

기본값은 "dlq"입니다.

payload

배열

옵션

API 엔드포인트로 전송되는 요청 본문을 사용자 지정할 수 있는 맞춤형 내부 파이프라인입니다. payload는 다음 표현식을 지원합니다.

  • $project

  • $addFields

  • $replaceRoot

  • $set

행동

$externalFunction 단계에서는 지정된 Amazon Web Services Lambda 리소스로 요청 보냅니다. 요청 이 동기식인 경우 지정된 필드 에 응답이 반환되고 파이프라인 처리 계속합니다. 요청 이 비동기식인 경우 파이프라인 응답을 기다리지 않고 처리 계속합니다.

동기 요청 실패하면 파이프라인의 오류 처리 동작은 onError 필드 에 의해 결정됩니다. 요청 이 비동기식인 경우 onError 필드 Amazon Web Services Lambda 함수 오류가 아닌 Amazon Web Services API 오류에만 적용됩니다. onError 필드 지정하지 않으면 기본값 동작은 영향을 받는 문서 데드 레터 큐로 보내는 것입니다.

onError 값
행동

"dlq"

영향을 받는 문서 데드 레터 큐 로 전송됩니다.

"ignore"

영향을 받는 문서 무시됩니다.

"fail"

스트림 프로세서는 오류로 종료됩니다.

파이프라인 문서 적절한 JSON 으로 변환할 수 없거나 구성된 파이프라인 최종 단계의 결과로 유효한 BSON 객체 생성하지 않는 경우 onError 설정에 관계없이 문서 데드 레터 큐 로 전송됩니다.

잘못된 표현식과 같이 $externalFunction 연산자 자체의 잘못된 구성으로 인해 발생하는 오류는 onError 동작을 트리거하다 하지 않습니다. 대신 스트림 프로세서가 실패하고 오류 메시지가 표시됩니다. Atlas Stream Processing 파이프라인 일시적인 오류로 인해 실패한 요청을 다시 시도합니다.

돌아가기

$merge

다음

$sessionWindow

이 페이지의 내용