Designing a Serverless Data Archiving Pipeline from Couchbase to Cloud Storage

최신 데이터 중심 애플리케이션에서는 규정 준수, 감사, 비용 최적화를 위해 기록 문서를 보관하는 것이 필수적입니다. 하지만 모든 데이터를 기본 운영 데이터베이스에 무기한 보관하는 것은 지속 불가능하고 비용이 많이 드는 경우가 많습니다.

이 블로그 게시물에서는 다음을 구축하는 방법을 안내합니다. 완전 서버리스 아카이브 파이프라인 에서 문서를 자동으로 이동하는 카우치베이스 에 Amazon S3를 사용하여 카우치베이스 이벤트, Amazon API 게이트웨이, SNS및 AWS 람다. 이 아키텍처는 다음을 활용하는 방법을 보여줍니다. 비동기식 디커플링 를 사용하여 복원력, 확장성 및 성능을 개선합니다.

이 튜토리얼을 마치면 수동 개입 없이도 Couchbase에서 문서 변경 또는 TTL 기반 만료에 대응하고 S3에 효율적으로 아카이브하는 강력한 엔드투엔드 솔루션을 갖추게 됩니다.

아키텍처 개요

아키텍처는 다음과 같습니다:

흐름:

1. Couchbase는 문서 조건(예: TTL 만료 또는 사용자 정의 아카이브: true 플래그).
2. 카우치베이스 이벤트 함수가 문서를 트리거하여 API 게이트웨이로 전송합니다.
3. API 게이트웨이는 문서를 SNS 토픽으로 전달합니다.
4. SNS는 주제에 가입된 람다 함수를 호출합니다.
5. Lambda는 날짜 기반 폴더 구조를 사용하여 전체 JSON 문서를 S3 버킷에 씁니다.

이 설정은 분리형이며 확장 가능하며 폴링이 필요하지 않습니다.

왜 아카이브에 카우치베이스 이벤트가 필요한가요?

카우치베이스 이벤트 는 문서 변경(생성, 업데이트, 삭제) 또는 만료에 대응하여 비즈니스 로직을 트리거하는 기본 방법을 제공합니다.

이벤트 기능을 사용하면 가능합니다:

- 특정 문서 유형 또는 필드(예 아카이브 === 참 및/또는 유형 === 로그)
- TTL 만료에 실시간으로 대응하기
- HTTP 호출을 통해 외부 서비스(예: AWS)로 데이터 푸시하기

카우치베이스 이벤트 함수 작성하기

다음은 문서 보관에 사용하는 Couchbase Eventing 함수의 간단한 예시입니다. 이 함수는 두 가지 주요 시나리오를 처리하는 로직을 구현합니다:

1. TTL 기반 아카이빙: 문서에 만료 설정하면 TTL 60초 전에 실행되는 타이머를 등록합니다. 타이머가 만료되면 문서 타이머 콜백 함수가 호출되고, 이 함수가 호출되면 게시(문서, 메타) 함수를 사용하여 문서를 보관할 수 있습니다.
2. 플래그 기반 아카이빙: 또는 문서에 다음과 같은 필드가 포함된 경우 아카이브: true를 호출하면 함수는 즉시 게시(문서, 메타) 를 클릭하여 문서를 보관합니다.

두 경우 모두 문서가 외부 API 게이트웨이로 전송되어 보관됩니다. API 응답 상태가 다음 중 하나에 해당하는 경우 200 또는 302를 누르면 문서가 소스 버킷에서 명시적으로 삭제되어 아카이브 워크플로우가 완료됩니다. 이는 온디맨드 또는 TTL 기반 자동화를 통해 문서를 아카이브할 수 있는 유연한 메커니즘을 제공합니다.

function OnUpdate(doc, meta) {
    if (doc.archive && doc.archive === true){
        log('Archiving document with ID:', meta.id);
        var status = publish(doc, meta);
        if (status === true) {
            delete src[meta.id];
        } else {
            log('Publish failed, document will not be deleted:', meta.id);
        }
    } else if (meta.expiration > 0){
        var nMinsPrior = new Date((meta.expiration - 60) * 1000);
        var currentTime = new Date().getTime();
        log('Time difference (ms): ', currentTime - nMinsPrior);
        if (currentTime > nMinsPrior) {
            log('Within 1 minute of TTL expiry, archiving:', meta.id);
            var publishStatus = publish(doc, meta);
        } else {
            log('Timer set for future archiving:', meta.id);
            createTimer(DocTimerCallback, nMinsPrior, meta.id, meta.id);
        }
    } else {
        log('No archiving conditions met for:', meta.id);
        return;
    }
}

function DocTimerCallback(context) {
    var doc = src[context];
    if (doc) {
        var meta = { id: context };
        var publishStatus = publish(doc, meta);
    } else {
        log('Timer callback failed: document not found for:', context);
    }
}

function publish(doc, meta) {
    try {
        var request = {
            path: 'archive',
            headers: {
                'Content-Type': 'application/json'
            },
            body: { ...doc, id: meta.id }
        };
        log("Sending request:", request);
        var response = curl('POST', archive2S3, request);
        if (response.status === 200 || response.status === 302) {
            log("Publish success for:", meta.id, " Response body:", response.body);
            return true;
        } else {
            log("Publish failed with status:", response.status, " Request body:", request);
            return false;
        }
    } catch (e) {
        log("Exception during publish:", e);
        return false;
    }
}

함수 온업데이트(doc, 메타) {

만약 (doc.아카이브 && doc.아카이브 === true){

로그('ID가 있는 문서 보관:', 메타.id);

var 상태 = 게시(doc, 메타);

만약 (상태 === true) {

삭제 src[메타.id];

} else {

로그('게시 실패, 문서가 삭제되지 않습니다:', 메타.id);

}

} else 만약 (메타.만료 > 0){

var nMinsPrior = new 날짜((메타.만료 - 60) * 1000);

var currentTime = new 날짜().getTime();

로그('시차(ms): ', currentTime - nMinsPrior);

만약 (currentTime > nMinsPrior) {

로그('TTL 만료 후 1분 이내, 보관:', 메타.id);

var 게시 상태 = 게시(doc, 메타);

} else {

로그('향후 보관을 위한 타이머 설정:', 메타.id);

createTimer(문서 타이머 콜백, nMinsPrior, 메타.id, 메타.id);

}

} else {

로그('다음에 대한 보관 조건이 충족되지 않음', 메타.id);

반환;

}

함수 문서 타이머 콜백(컨텍스트) {

var doc = src[컨텍스트];

만약 (doc) {

var 메타 = { id: 컨텍스트 };

var 게시 상태 = 게시(doc, 메타);

} else {

로그('타이머 콜백 실패: 문서를 찾을 수 없음:', 컨텍스트);

}

함수 게시(doc, 메타) {

시도 {

var 요청 = {

경로: '아카이브',

헤더: {

'콘텐츠 유형': 'application/json'

body: { ...doc, id: 메타.id }

};

로그("요청 보내기:", 요청);

var 응답 = curl('POST', archive2S3, 요청);

만약 (응답.상태 === 200 || 응답.상태 === 302) {

로그("게시 성공 대상:", 메타.id, " 응답 본문:", 응답.body);

반환 true;

} else {

로그("게시 실패 상태:", 응답.상태, " 요청 본문:", 요청);

반환 false;

}

} catch (e) {

로그("게시 중 예외:", e);

반환 false;

}

참고: 성능상의 이유로 모든 주석 처리된 로그() 문으로 표시됩니다. 이러한 로그는 주로 디버깅 및 개발 목적으로 포함되었습니다. 프로덕션 환경에서 과도한 로깅은 성능에 영향을 미치고 로그 스토리지 비용을 증가시킬 수 있습니다.

이벤트 함수를 만들면서 설정과 바인딩을 정의한 방법은 다음과 같습니다.

히트 다음 를 사용하여 바인딩을 생성합니다. 여기에서 API 게이트웨이의 엔드포인트를 별칭에 바인딩합니다. archive2S3 소스 버킷을 별칭으로 사용하며 src. 소스 버킷에서 데이터를 제거하기 위해 읽기/쓰기 권한을 사용했다는 점에 유의하세요.

히트 다음 를 다시 클릭하고 위의 JS 함수를 창에 복사/붙여넣습니다. 저장. 이 시점에서 함수는 저장되지만 배포되지는 않습니다. 점 세 개를 누르고 배포 옵션을 사용하여 이벤트 함수를 실행합니다. 함수가 실행되면 다음과 같이 표시됩니다.

S3에 아카이브하는 람다 함수 구축하기

람다 함수는 SNS 메시지를 소비하고 전체 JSON을 날짜별로 정리하여 S3 버킷에 보관합니다.

람다 코드 예시

import boto3
import json
from datetime import datetime

s3 = boto3.client('s3')
bucket_name = 'your-s3-archive-bucket'

def lambda_handler(event, context):
    for record in event['Records']:
        msg = json.loads(record['Sns']['Message'])
        doc_id = msg['id']
        content = msg
        now = datetime.utcnow()
        folder = f"{now.year}/{now.month}/{now.day}"
        key = f"{folder}/{doc_id}.json"
        s3.put_object(
            Bucket=bucket_name,
            Key=key,
            Body=json.dumps(content),
            ContentType='application/json'
        )

    return {
        'statusCode': 200,
        'body': 'Archived successfully.'
    }

가져오기 boto3

가져오기 json

에서 날짜 시간 가져오기 날짜 시간

s3 = boto3.클라이언트('s3')

버킷_이름 = 'your-s3-아카이브-버킷'

def 람다_핸들러(이벤트, 컨텍스트):

에 대한 기록 in 이벤트['Records']:

msg = json.로드(기록['Sns']['메시지'])

doc_id = msg['id']

콘텐츠 = msg

지금 = 날짜 시간.utcnow()

폴더 = f"{now.year}/{now.month}/{now.day}"

키 = f"{folder}/{doc_id}.json"

s3.put_object(

버킷=버킷_이름,

키=키,

본문=json.덤프(콘텐츠),

콘텐츠 유형='application/json'

)

반환 {

'statusCode': 200,

'body': '성공적으로 보관되었습니다.

}

그 결과 다음과 같은 S3 개체가 생성됩니다:

s3://your-s3-archive-bucket/2025/6/29/log123.json

1	s3://your-s3-archive-bucket/2025/6/29/log123.json

SNS를 사용하여 아카이브 트리거 분리하기

SNS(간편 알림 서비스) 를 사용하면 여러 서비스가 동일한 메시지를 수신할 수 있습니다. 여기서는 아카이브 요청을 람다 함수에 전달합니다.

단계:

- 예를 들어 주제를 만듭니다, 아카이브 트리거 주제
- API 게이트웨이가 IAM을 통해 게시하도록 허용하기
- 람다 함수 구독

권한 수정

신뢰 및 액세스 정책을 통해 API 게이트웨이가 SNS 토픽에 게시할 수 있는지 확인합니다:

다음에 대한 신뢰 정책 `SNS접속` 역할:

{
  "Effect": "Allow",
  "Principal": {
    "Service": "apigateway.amazonaws.com"
  },
  "Action": "sts:AssumeRole"
}

{

"효과": "허용",

"교장": {

"서비스": "apigateway.amazonaws.com"

"액션": "sts:가정 역할"

}

역할에 대한 액세스 정책:

{
  "Effect": "Allow",
  "Action": "sns:Publish",
  "Resource": "arn:aws:sns:us-east-1:account-id:ArchiveTriggerTopic"
}

{

"효과": "허용",

"액션": "SNS:게시",

"리소스": "arn:aws:sns:us-east-1:account-id:ArchiveTriggerTopic"

}

아카이브 요청을 수락하도록 API 게이트웨이 설정하기

API 게이트웨이는 아카이브 요청을 받아 SNS로 전달하는 퍼블릭 엔드포인트 역할을 합니다.

주요 단계:

- 만들기 REST API API 게이트웨이에서.

제공된 옵션에서 다음을 선택합니다. REST API 와의 통합을 제공하므로 SNS 서비스.

제공 API 이름 을 클릭하고 API 엔드포인트 유형 as 지역. Hit API 만들기 버튼을 클릭합니다.

- 설정 POST /아카이브 경로.

다음 리소스 페이지에서 리소스를 생성하려면 리소스 만들기 버튼을 클릭합니다.

제공 리소스 이름. 리소스 이름을 다음과 같이 호출합니다. 아카이브.

히트 리소스 만들기 버튼을 클릭합니다. 다음 페이지의 방법 창 히트 생성 방법 버튼을 클릭합니다. 이렇게 하면 여러 가지 설정을 POST 메서드에 매핑하고 SNS 어떤 서비스 AWS 지역 실행 중이면 IAM 역할 SNS 토픽에 게시하는 데 필요한 권한을 가지고 있습니다.

페이지를 아래로 스크롤하면 다음을 찾을 수 있습니다. URL 쿼리 문자열 매개변수 섹션을 참조하세요. 여기서는 매개변수 이름을 사용하여 SNS 토픽의 ARN을 매핑하겠습니다. TopicArn. 또한 메시지 에 method.request.body를 클릭하면 JSON 문서의 전체 페이로드가 포함됩니다.

마지막으로 저장 버튼을 클릭합니다.

방금 배포한 것을 축하드립니다. API 게이트웨이 에 JSON 문서를 게시할 수 있는 SNS 토픽를 트리거하여 궁극적으로 Lambda 에 작성하려면 S3.

엔드투엔드 흐름 테스트

두 가지 방법으로 파이프라인을 테스트할 수 있습니다:

먼저 API POST 메서드를 테스트합니다.

- 히트 테스트 탭에서 간단한 JSON을 제출하고 id 필드를 필수로 설정합니다.

를 누르면 테스트 버튼을 눌러 로그 추적에 오류가 없고 응답 상태가 200인지 확인합니다. 이 시점에서 API 엔드포인트가 정상적으로 작동합니다. 다음으로 curl에서 이 서비스를 테스트해 보겠습니다.

컬 또는 포스트맨에서

curl -X POST https://your-api-gateway-url/archive \
  -H "Content-Type: application/json" \
  -d '{
        "id": "hotel::10025",
        "type": "Hotel",
        "message": "Archiving via curl",
        "archive": true
      }'

curl -X POST https://your-api-gateway-url/archive \

-H "콘텐츠 유형: 애플리케이션/json" \

-d '{

"id": "hotel::10025",

"유형": "호텔",

"메시지": "컬을 통한 보관",

"archive": true

트리거 후 S3 버킷에 해당 오브젝트가 생성되었는지 확인합니다.

쿼리 워크벤치를 사용하여 카펠라에서

에서 설정을 테스트하려면 카펠라를 사용하여 문서를 삽입하고 TTL 값 120초.

UPSERT INTO bulk.data.source (KEY, VALUE)
VALUES ("test::001", {"type": "test", "field": "value"}, {"expiration": 2*60});

UPSERT INTO 대량.데이터.출처 (KEY, VALUE)

가치 ("test::001", {"type": "test", "field": "value"}, {"만료": 2*60});

쿼리 워크벤치에서 위의 SQL 명령을 실행한 다음 문서가 구성된 S3 버킷 대략 만료 60초 전이벤트 함수는 타이머를 트리거하도록 설정하므로 TTL 1분 전.

문제 해결

다음은 몇 가지 일반적인 문제와 해결 방법입니다:

유효성 검사 오류: 메시지는 널이 아니어야 합니다.

- 이는 일반적으로 메시지 SNS로 전송된 필드가 비어 있습니다.
- 다음 사항을 확인하십시오. API 게이트웨이 매핑 템플릿 는 본문을 올바르게 추출하고 있습니다.

API 게이트웨이에 역할을 맡을 권한이 없습니다.

- IAM 역할이 올바른지 확인합니다. 신뢰 정책.
- 이 역할은 다음을 허용해야 합니다. apigateway.amazonaws.com 서비스를 가정합니다.

API 요청의 잘못된 콘텐츠 유형

- API 게이트웨이는 콘텐츠 유형이 다음과 같은 경우에만 매핑 템플릿을 적용합니다. application/json.
- 카우치베이스 이벤트 함수(또는 포스트맨)가 이 헤더를 설정했는지 확인하세요.

이스케이프되거나 변형된 JSON을 수신하는 SNS

- 사용 중인 $util.escapeJavaScript($input.body) 를 매핑 템플릿에 추가합니다.
- 이스케이프가 잘못되면 다운스트림 람다 구문 분석에 문제가 발생할 수 있습니다.

Lambda를 검사하기 위한 CloudWatch 로그

- 람다 함수의 실행 추적을 모니터링하여 모든 것이 예상대로 실행되었는지 확인합니다.

개선 사항 및 모범 사례

- 사용 환경 변수 에서 S3 버킷 이름과 지역에 대한 람다를 입력합니다.
- 사용 S3 서버 측 암호화 (SSE-S3 또는 SSE-KMS)를 준수합니다.
- 켜기 S3 버전 관리 를 사용하여 과거 사본을 보존합니다.
- 추가 CloudWatch 알람 람다 오류 또는 API 게이트웨이 5XX를 확인합니다.
- 사용 SNS 팬아웃 를 사용하여 추가 소비자(예: 키네시스, 기타 람다)에게 알릴 수 있습니다.
- SNS를 다음과 같이 대체하는 것이 좋습니다. 직접 람다 통합 소비자가 한 명뿐이고 간단한 권한을 원하는 경우.

결론

이 블로그 게시물에서는 다음을 사용하여 강력한 실시간 문서 보관 파이프라인을 구축했습니다:

- 보관 가능한 문서를 감지하는 Couchbase Eventing
- 공용 엔드포인트를 노출하는 API 게이트웨이
- 생산자와 소비자를 분리하는 SNS
- 문서를 처리하고 S3에 저장하는 Lambda

이 아키텍처는 완전히 서버리스이며 손쉽게 확장할 수 있고 보존, 규정 준수 또는 분석을 위해 기록 데이터를 오프로드하는 비용 효율적인 방법입니다.

리소스

이 파이프라인에 사용되는 기술에 대해 더 자세히 알아보고 지식을 넓히는 데 도움이 되는 몇 가지 유용한 리소스를 소개합니다:

Anuj Sahni, 솔루션 아키텍처 관리자, Couchbase

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

모든 게시물

카우치베이스에서 클라우드 스토리지로 서버리스 데이터 아카이빙 파이프라인 설계하기

아키텍처 개요

왜 아카이브에 카우치베이스 이벤트가 필요한가요?

카우치베이스 이벤트 함수 작성하기

S3에 아카이브하는 람다 함수 구축하기

람다 코드 예시

SNS를 사용하여 아카이브 트리거 분리하기

다음에 대한 신뢰 정책 SNS접속 역할:

역할에 대한 액세스 정책:

아카이브 요청을 수락하도록 API 게이트웨이 설정하기

엔드투엔드 흐름 테스트

먼저 API POST 메서드를 테스트합니다.

컬 또는 포스트맨에서

쿼리 워크벤치를 사용하여 카펠라에서

문제 해결

유효성 검사 오류: 메시지는 널이 아니어야 합니다.

API 게이트웨이에 역할을 맡을 권한이 없습니다.

API 요청의 잘못된 콘텐츠 유형

이스케이프되거나 변형된 JSON을 수신하는 SNS

Lambda를 검사하기 위한 CloudWatch 로그

개선 사항 및 모범 사례

결론

리소스

작성자

게시자 Anuj Sahni, 솔루션 아키텍처 관리자, Couchbase

댓글 남기기 응답 취소

다음에 대한 신뢰 정책 `SNS접속` 역할: