튜토리얼: Lambda를 Kinesis Data Streams와 함께 사용하기

튜토리얼: Lambda를 Kinesis Data Streams와 함께 사용하기 (Tutorial: Using Lambda with Kinesis Data Streams)

이 튜토리얼에서는 Amazon Kinesis 데이터 스트림의 이벤트를 소비하는 Lambda 함수를 만들어요.

커스텀 앱이 스트림에 레코드를 써요.
AWS Lambda가 스트림을 폴링하고, 스트림에서 새 레코드를 감지하면 Lambda 함수를 호출해요.
AWS Lambda는 Lambda 함수를 만들 때 지정한 실행 역할을 수임(assume)해 Lambda 함수를 실행해요.

전제 조건

아직 AWS Command Line Interface를 설치하지 않았다면 AWS CLI 최신 버전 설치·업데이트하기의 단계를 따라 설치해요.

튜토리얼은 명령을 실행하기 위해 명령줄 터미널이나 셸이 필요해요. Linux와 macOS에서는 원하는 셸과 패키지 매니저를 사용해요.

참고 Windows에서는 Lambda에 자주 쓰는 일부 Bash CLI 명령(예: zip)이 운영체제 내장 터미널에서 지원되지 않아요. Windows와 통합된 Ubuntu와 Bash를 얻으려면 Windows Subsystem for Linux를 설치하세요.

출처: AWS Lambda 개발자 안내서

본문

실행 역할 만들기

함수에게 AWS 리소스에 접근할 권한을 주는 실행 역할(execution role)을 만들어요.

실행 역할을 만들려면

IAM 콘솔의 roles 페이지를 열어요.
Create role을 선택해요.
다음 속성으로 역할을 만들어요. Trusted entity — AWS Lambda. Permissions — AWSLambdaKinesisExecutionRole. Role name — lambda-kinesis-role.

AWSLambdaKinesisExecutionRole 정책에는 함수가 Kinesis에서 항목을 읽고 CloudWatch Logs에 로그를 쓸 때 필요한 권한이 들어 있어요.

함수 만들기

Kinesis 메시지를 처리하는 Lambda 함수를 만들어요. 함수 코드는 Kinesis 레코드의 이벤트 ID와 이벤트 데이터를 CloudWatch Logs에 기록해요.

이 튜토리얼은 Node.js 24 런타임을 사용하지만, 다른 런타임 언어의 예시 코드도 제공해요. 관심 있는 런타임의 코드를 보려면 아래 상자의 탭을 선택하세요. 이 단계에서 사용할 JavaScript 코드는 JavaScript 탭의 첫 번째 예시에 있어요.

예시 — JavaScript (SDK for JavaScript v3)

Kinesis 이벤트를 Lambda로 사용하기 (JavaScript):

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: Apache-2.0
exports.handler = async (event, context) => {
  for (const record of event.Records) {
    try {
      console.log(`Processed Kinesis Event - EventID: ${record.eventID}`);
      const recordData = await getRecordDataAsync(record.kinesis);
      console.log(`Record Data: ${recordData}`);
      // TODO: Do interesting work based on the new data
    } catch (err) {
      console.error(`An error occurred ${err}`);
      throw err;
    }
  }
  console.log(`Successfully processed ${event.Records.length} records.`);
};

async function getRecordDataAsync(payload) {
  var data = Buffer.from(payload.data, "base64").toString("utf-8");
  await Promise.resolve(1); //Placeholder for actual async work
  return data;
}

참고 GitHub에 더 많은 내용이 있어요. Serverless examples 저장소에서 전체 예시를 찾고 설정·실행 방법을 알아보세요. 이 튜토리얼은 Node.js 런타임을 사용하지만, .NET, Go, Java, TypeScript, PHP, Python, Ruby, Rust용 예시 코드도 원본 문서에서 각각의 SDK 탭에서 확인할 수 있어요.

함수를 만들려면

프로젝트용 디렉터리를 만들고 그 디렉터리로 이동해요.

mkdir kinesis-tutorial
cd kinesis-tutorial

샘플 JavaScript 코드를 index.js라는 새 파일에 복사해요.
배포 패키지를 만들어요.

zip function.zip index.js

create-function 명령으로 Lambda 함수를 만들어요.

aws lambda create-function --function-name ProcessKinesisRecords \
--zip-file fileb://function.zip --handler index.handler --runtime nodejs24.x \
--role arn:aws:iam::111122223333:role/lambda-kinesis-role

Lambda 함수 테스트하기

invoke AWS Lambda CLI 명령과 샘플 Kinesis 이벤트로 Lambda 함수를 수동으로 호출해요.

Lambda 함수를 테스트하려면

다음 JSON을 파일에 복사하고 input.txt로 저장해요.

{
  "Records": [
    {
      "kinesis": {
        "kinesisSchemaVersion": "1.0",
        "partitionKey": "1",
        "sequenceNumber": "49590338271490256608559692538361571095921575989136588898",
        "data": "SGVsbG8sIHRoaXMgaXMgYSB0ZXN0Lg==",
        "approximateArrivalTimestamp": 1545084650.987
      },
      "eventSource": "aws:kinesis",
      "eventVersion": "1.0",
      "eventID": "shardId-000000000006:***",
      "eventName": "aws:kinesis:record",
      "invokeIdentityArn": "arn:aws:iam::111122223333:role/lambda-kinesis-role",
      "awsRegion": "us-east-2",
      "eventSourceARN": "arn:aws:kinesis:us-east-2:111122223333:stream/lambda-stream"
    }
  ]
}

invoke 명령으로 이벤트를 함수에 보내요.

aws lambda invoke --function-name ProcessKinesisRecords \
--cli-binary-format raw-in-base64-out \
--payload file://input.txt outputfile.txt

cli-binary-format 옵션은 AWS CLI version 2를 쓸 때 필요해요. 이걸 기본 설정으로 만들려면 aws configure set cli-binary-format raw-in-base64-out을 실행하세요. 자세한 내용은 AWS Command Line Interface User Guide for Version 2의 AWS CLI supported global command line options을 참고하세요. 응답은 out.txt에 저장돼요.

Kinesis 스트림 만들기

create-stream 명령으로 스트림을 만들어요.

aws kinesis create-stream --stream-name lambda-stream --shard-count 1

다음 describe-stream 명령으로 스트림 ARN을 가져와요.

aws kinesis describe-stream --stream-name lambda-stream

다음 출력이 보여야 해요.

{
  "StreamDescription": {
    "Shards": [
      {
        "ShardId": "shardId-000000000000",
        "HashKeyRange": {
          "StartingHashKey": "0",
          "EndingHashKey": "340282366920746074317682119384634633455"
        },
        "SequenceNumberRange": {
          "StartingSequenceNumber": "49591073947768692513481539594623130411957558361251844610"
        }
      }
    ],
    "StreamARN": "arn:aws:kinesis:us-east-1:111122223333:stream/lambda-stream",
    "StreamName": "lambda-stream",
    "StreamStatus": "ACTIVE",
    "RetentionPeriodHours": 24,
    "EnhancedMonitoring": [
      {
        "ShardLevelMetrics": []
      }
    ],
    "EncryptionType": "NONE",
    "KeyId": null,
    "StreamCreationTimestamp": 1544828156.0
  }
}

다음 단계에서 스트림을 Lambda 함수와 연결하려고 스트림 ARN을 사용해요.

AWS Lambda에 이벤트 소스 추가하기

다음 AWS CLI add-event-source 명령을 실행해요.

aws lambda create-event-source-mapping --function-name ProcessKinesisRecords \
--event-source arn:aws:kinesis:us-east-1:111122223333:stream/lambda-stream \
--batch-size 100 --starting-position LATEST

나중에 쓰려고 매핑 ID를 기억해 두세요. list-event-source-mappings 명령을 실행해 이벤트 소스 매핑 목록을 얻을 수 있어요.

aws lambda list-event-source-mappings --function-name ProcessKinesisRecords \
--event-source arn:aws:kinesis:us-east-1:111122223333:stream/lambda-stream

응답에서 상태 값이 enabled인지 확인할 수 있어요. 이벤트 소스 매핑은 레코드를 잃지 않고 폴링을 일시 중지하도록 비활성화할 수 있어요.

설정 테스트하기

이벤트 소스 매핑을 테스트하려면 Kinesis 스트림에 이벤트 레코드를 추가해요. --data 값은 CLI가 Kinesis로 보내기 전에 base64로 인코딩하는 문자열이에요. 같은 명령을 여러 번 실행해 스트림에 여러 레코드를 추가할 수 있어요.

aws kinesis put-record --stream-name lambda-stream --partition-key 1 \
--data "Hello, this is a test."

Lambda는 실행 역할로 스트림에서 레코드를 읽어요. 그런 다음 Lambda 함수를 호출하면서 레코드 배치를 전달해요. 함수는 각 레코드의 데이터를 디코딩하고 로그로 기록해 CloudWatch 콘솔로 출력을 보내요.

리소스 정리하기

이제 이 튜토리얼을 위해 만든 리소스를 계속 유지하고 싶지 않다면 삭제할 수 있어요. 더 이상 사용하지 않는 AWS 리소스를 삭제하면 AWS 계정에 불필요한 요금이 부과되는 것을 막을 수 있어요.

실행 역할을 삭제하려면

IAM 콘솔의 Roles 페이지를 열어요.
만든 실행 역할을 선택해요.
Delete를 선택해요.
텍스트 입력 필드에 역할 이름을 입력하고 Delete를 선택해요.

Lambda 함수를 삭제하려면

Lambda 콘솔의 Functions 페이지를 열어요.
만든 함수를 선택해요.
Actions, Delete를 선택해요.
텍스트 입력 필드에 confirm을 입력하고 Delete를 선택해요.

Kinesis 스트림을 삭제하려면

AWS Management Console에 로그인하고 https://console.aws.amazon.com/kinesis에서 Kinesis 콘솔을 열어요.
만든 스트림을 선택해요.
Actions, Delete를 선택해요.
텍스트 입력 필드에 delete를 입력해요.
Delete를 선택해요.

튜토리얼: Lambda를 Kinesis Data Streams와 함께 사용하기