DynamoDB Stream이란?

DynamoDB Stream은 DynamoDB에 대한 항목 추가, 변경 및 삭제를 이벤트로 감지할 수 있는 기능입니다. DynamoDB Stream을 사용하는 것으로, 항목이 추가·변경되었을 때에 푸시 통지를 날리거나 하는 구현이 가능하게 됩니다. 이번에는 그 이벤트를 Lambda에서 받고 싶습니다.

참고 : DynamoDB 스트림을 사용하여 테이블 활동 캡처

DynamoDB 측에서 Stream 사용

먼저 DynamoDB 개요 탭의 "스트림 관리"에서 Stream을 활성화합니다. 새로 고침된 이미지를 선택하면 Lambda가 항목을 변경할 때 항목의 Before&After를 모두 받을 수 있습니다.

DynamoDB Stream을 받기 위해 Lambda에 필요한 권한

Lambda가 DynamoDB Stream을 받으려면 읽으려는 테이블 스트림에 대해 다음 네 가지 권한을 허용해야 합니다. 스트림의 ARN은, arn:aws:dynamodb:ap-northeast-1:XXXXXXXXX:table/テーブル名/stream/* 로 하는 것으로 모든 스트림을 받을 수가 있습니다.

GetShardIterator

GetRecords

ListStream

DescribeStream

Lambda 측에서 DynamoDB Stream 트리거 추가

Lambda 페이지의 "왼쪽에서 트리거 추가"라는 곳에서 DynamoDB를 선택하면 설정 화면이 나옵니다.


설정은 Stream을 수신할 테이블과 배치 크기를 지정합니다. 시작 위치는 최신과 수평을 선택할 수 있습니다. 여기에서 최신을 선택합니다. 왼쪽 하단 추가를 클릭 한 후 변경 사항을 저장하면 트리거로 추가됩니다.

Lambda 함수의 내용의 병아리

event.Records가 배열로 이벤트를 받고 있기 때문에 forEach에서 하나 하나 처리하고 있습니다. eventName에 "INSERT", "MODIFY"또는 "REMOVE"가 들어있어 변경의 종류를 알려줍니다.

index.js

exports.handler = (event, context, callback) => {

    event.Records.forEach((record) => {
        console.log('イベント種別:', record.eventName);
        console.log('DynamoDB Record: %j', record.dynamodb);

        if(record.eventName == 'INSERT'){
            //項目が追加された時の処理
            const newItem = record.dynamodb.NewImage;

        }else if(record.eventName == 'MODIFY'){
            //項目が変更された時の処理
            const oldItem = record.dynamodb.OldImage;//変更前
            const newItem = record.dynamodb.NewImage;//変更後

        }else if(record.eventName == 'REMOVE'){
            //項目が削除された時の処理
            const deletedItem = record.dynamodb.OldImage

        }else{

        }
    });

};

결과

DynamoDB의 콘솔 화면에서 적절하게 항목을 추가, 편집, 삭제해 봅시다. 아래의 로그는 항목을 편집 (이름을 야마다 타로 → 사토 타로로 변경)했을 때의 로그입니다. NewImage와 OldImage 안에 Before&After가 들어 있는 것을 알 수 있다고 생각합니다. S라든지 N라든지는 데이터형을 나타내고 있습니다.

{
    "ApproximateCreationDateTime": 1518610740,
    "Keys": {
        "id": {
            "S": "sample"
        },
        "create_at": {
            "N": "1518534000"
        }
    },
    "NewImage": {
        "name": {
            "S": "佐藤太郎"
        },
        "id": {
            "S": "sample"
        },
        "create_at": {
            "N": "1518534000"
        }
    },
    "OldImage": {
        "name": {
            "S": "山田太郎"
        },
        "id": {
            "S": "sample"
        },
        "create_at": {
            "N": "1518534000"
        }
    },
    "SequenceNumber": "24000000000007967212728",
    "SizeBytes": 98,
    "StreamViewType": "NEW_AND_OLD_IMAGES"
}