기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다. # 자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리 이 자습서의 시나리오에서는 스트림에 주식 거래를 가져와 데이터 스트림에 대한 계산을 수행하는 간단한 Amazon Kinesis Data Streams 애플리케이션을 작성합니다. 레코드의 스트림을 Kinesis Data Streams에 전송하고, 거의 실시간으로 레코드를 사용하고 처리하는 애플리케이션을 구현하는 방법에 대해 알아봅니다. **중요** 스트림을 생성한 후에는 Kinesis Data Streams가 AWS 프리 티어에 적합하지 않기 때문에 계정에 Kinesis Data Streams 사용에 대한 일반 요금이 발생합니다. 소비자 애플리케이션이 시작된 후 Amazon DynamoDB 사용량에 대해 일반 요금도 발생합니다. 소비자 애플리케이션은 DynamoDB를 사용하여 처리 상태를 추적합니다. 이 애플리케이션을 완료하면 AWS 리소스를 삭제하여 요금 발생을 중지하세요. 자세한 내용은 [리소스 정리](tutorial-stock-data-kplkcl-finish.md) 단원을 참조하십시오. 이 코드는 실제 주식 시장 데이터에는 액세스하지 않지만, 대신 주식 거래의 스트림을 시뮬레이션합니다. 2015년 2월 현재 시가 총액 상위 25개 주식에 대한 실제 시장 데이터의 시작점이 있는 임의의 주식 거래 생성기를 사용하여 이를 수행합니다. 주식 거래의 실시간 스트림에 액세스할 수 있는 경우 스트림에서 유용하고 시기 적절한 통계를 추출하고 싶을 때도 있습니다. 예를 들어, 마지막 5분 이내에 구매한 가장 인기 있는 주식을 결정하는 슬라이딩 윈도우 분석을 수행하려고 할 수 있습니다. 또는 너무 많은 판매 주문(즉, 너무 많은 공유)이 있을 때마다 알림을 원할 수도 있습니다. 이 시리즈의 코드를 확장하여 이러한 기능을 제공할 수 있습니다. 데스크톱 또는 랩톱 컴퓨터에서 이 자습서의 단계를 수행하고, 동일한 시스템에서 생산자 코드와 소비자 코드를 둘 다 실행하거나 Amazon Elastic Compute Cloud(Amazon EC2)와 같은 정의된 요건을 지원하는 모든 플랫폼을 실행할 수 있습니다. 표시된 예제는 미국 서부(오레곤) 리전을 사용하지만 이 예제는 [Kinesis Data Streams를 지원하는 모든AWS 리전](https://docs.aws.amazon.com/general/latest/gr/rande.html#ak_region)에 적용됩니다. **Topics** + [사전 조건 완료](tutorial-stock-data-kplkcl-begin.md) + [데이터 스트림 생성](tutorial-stock-data-kplkcl-create-stream.md) + [IAM 정책 및 사용자 생성](tutorial-stock-data-kplkcl-iam.md) + [구현 코드 다운로드 및 빌드](tutorial-stock-data-kplkcl-download.md) + [생산자 구현](tutorial-stock-data-kplkcl-producer.md) + [소비자 구현](tutorial-stock-data-kplkcl-consumer.md) + [(선택 사항) 소비자 확장](tutorial-stock-data-kplkcl-consumer-extension.md) + [리소스 정리](tutorial-stock-data-kplkcl-finish.md) # 사전 조건 완료 다음은 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리[자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)](tutorial-stock-data-kplkcl.md) 단원을 완료하는 데 필요한 요구 사항입니다. ## Amazon Web Services 계정 생성 및 사용 시작에 앞서 [Amazon Kinesis Data Streams 용어 및 개념](key-concepts.md)에 기술된 개념을 잘 알아야 합니다. 특히 스트림, 샤드, 생산자 및 소비자에 유의하십시오. [자습서: Kinesis Data Streams AWS CLI 용 설치 및 구성](kinesis-tutorial-cli-installation.md) 단원의 내용을 숙지하면 도움이 됩니다. 에 액세스하려면 AWS 계정과 웹 브라우저가 필요합니다 AWS Management Console. 콘솔 액세스의 경우 IAM 사용자 이름과 암호를 사용하여 IAM 로그인 페이지에서 [AWS Management Console](https://console.aws.amazon.com/console/home)에 로그인합니다. 프로그래밍 방식 액세스 및 장기 자격 증명의 대안을 포함한 AWS 보안 자격 증명에 대한 자세한 내용은 *IAM 사용 설명서*의 [AWS 보안 자격 증명을](https://docs.aws.amazon.com/IAM/latest/UserGuide/security-creds.html) 참조하세요. 에 로그인하는 방법에 대한 자세한 내용은 *AWS Sign-In 사용 설명서*의에 로그인하는 방법을 AWS 계정참조하세요. [AWS](https://docs.aws.amazon.com/signin/latest/userguide/how-to-sign-in.html) IAM 및 보안 키 설정 지침에 대한 자세한 내용은 [IAM 사용자 생성](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/get-set-up-for-amazon-ec2.html#create-an-iam-user)을 참조하세요. ## 시스템 소프트웨어 요구 사항 충족 애플리케이션을 실행하는 데 사용된 시스템에는 Java 7 이상이 설치되어 있어야 합니다. 최신 JDK(Java Development Kit)를 다운로드하고 설치하려면 [Oracle의 Java SE 설치 사이트](http://www.oracle.com/technetwork/java/javase/downloads/index.html)로 이동하십시오. [Eclipse](https://www.eclipse.org/downloads/)와 같이 Java IDE가 있는 경우 소스 코드를 열고 편집, 빌드 및 실행할 수 있습니다. 최신 [AWS SDK for Java](https://aws.amazon.com/sdk-for-java/) 버전이 필요합니다. IDE로 Eclipse를 사용하는 경우 [AWS Toolkit for Eclipse](https://aws.amazon.com/eclipse/)를 대신 설치할 수 있습니다. 소비자 애플리케이션에는 [Kinesis Client Library(Java)](https://github.com/awslabs/amazon-kinesis-client)의 GitHub에서 얻을 수 있는 Kinesis Client Library(KCL) 버전 1.2.1 이상이 필요합니다. ## 다음 단계 [데이터 스트림 생성](tutorial-stock-data-kplkcl-create-stream.md) # 데이터 스트림 생성 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리[자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)](tutorial-stock-data-kplkcl.md)의 첫 단계에서는 후속 단계에서 사용할 스트림을 생성합니다. **스트림을 만들려면** 1. 에 로그인 AWS Management Console 하고 [https://console.aws.amazon.com/kinesis](https://console.aws.amazon.com/kinesis) Kinesis 콘솔을 엽니다. 1. 탐색 창에서 **Data Streams(데이터 스트림)**를 선택합니다. 1. 탐색 모음에서 리전 선택기를 확장하고 리전을 선택합니다. 1. **Create Kinesis stream(Kinesis 스트림 생성)**을 선택합니다. 1. 스트림 이름을 입력합니다(예: **StockTradeStream**). 1. 샤드 수에는 **1**을 입력하고, **필요한 샤드 수 추정**은 축소된 상태로 둡니다. 1. **Create Kinesis stream(Kinesis 스트림 생성)**을 선택합니다. 스트림이 생성되는 동안 **Kinesis 스트림** 목록 페이지에서 스트림 상태는 `CREATING`입니다. 스트림을 사용할 준비가 되면 상태가 `ACTIVE`(활성)로 변경됩니다. 스트림 명칭을 선택합니다. 다음에 나타나는 페이지의 **Details(세부 정보)** 탭에는 스트림 구성 요약이 표시됩니다. **Monitoring(모니터링)** 섹션에는 스트림에 대한 모니터링 정보가 표시됩니다. ## 샤드에 대한 추가 정보 이 자습서가 아니라 실제로 Kinesis Data Streams를 사용하기 시작할 경우 스트림 생성 프로세스를 더 신중하게 계획해야 할 수 있습니다. 샤드를 프로비저닝할 때 예상되는 최대 수요를 계획해야 합니다. 이 시나리오를 예제로 사용하면, 미국 주식 시장 거래 트래픽이 낮(동부 시간) 동안 최대가 되며 수요 예상은 해당 시간대에서 샘플링되어야 합니다. 그런 다음 최대 예상 수요에 대한 프로비저닝을 선택하거나 수요 변동에 따라 스트림을 확장 및 축소합니다. *샤드*는 처리 용량의 단위입니다. **Kinesis 스트림 생성** 페이지에서 **필요한 샤드 수 추정**을 확장합니다. 다음 지침에 따라 평균 레코드 크기, 초당 작성된 최대 레코드 및 사용하는 애플리케이션의 수를 입력합니다. **평균 레코드 크기** 계산된 평균 레코드 크기의 추정입니다. 이 값을 모르는 경우 이 값에 대해 예상 최대 레코드 크기를 사용하십시오. **작성된 최대 레코드** 데이터를 제공하는 개체의 수와 각 개체가 생성한 대략적인 초당 레코드 수를 고려하세요. 예를 들어, 20개의 거래 서버에서 주식 거래 데이터를 가져오고 각각 초당 250개의 거래를 생성하는 경우 초당 총 거래(레코드) 수는 초당 5000개입니다. **사용하는 애플리케이션 수** 다른 방식으로 스트림을 처리하고 다른 출력을 생성하기 위해 스트림에서 독립적으로 읽는 애플리케이션의 수입니다. 각 애플리케이션에는 다른 시스템에서 실행(즉, 클러스터에서 실행)되는 여러 인스턴스가 있을 수 있으므로 대량의 스트림을 유지할 수 있습니다. 표시된 예상 샤드 수가 현재 샤드 제한을 초과하는 경우 해당 샤드 수가 있는 스트림을 생성하기 전에 해당 제한의 증가 요청을 제출해야 할 수도 있습니다. 샤드 제한 증가를 요청하려면 [Kinesis Data Streams 제한 양식](https://console.aws.amazon.com/support/home#/case/create?issueType=service-limit-increase&limitType=service-code-kinesis)을 사용하세요. 스트림 및 샤드에 대한 자세한 내용은 [Kinesis 데이터 스트림 생성 및 관리](working-with-streams.md) 섹션을 참조하세요. ## 다음 단계 [IAM 정책 및 사용자 생성](tutorial-stock-data-kplkcl-iam.md) # IAM 정책 및 사용자 생성 보안 모범 사례에 AWS 따라 세분화된 권한을 사용하여 다양한 리소스에 대한 액세스를 제어해야 합니다. AWS Identity and Access Management (IAM)을 사용하면에서 사용자 및 사용자 권한을 관리할 수 있습니다 AWS. [IAM 정책](https://docs.aws.amazon.com/IAM/latest/UserGuide/PoliciesOverview.html)에는 허용된 작업과 작업이 적용되는 리소스가 명시적으로 나열됩니다. 다음은 Kinesis Data Streams 생산자 및 소비자에 대해 일반적으로 필요한 최소 권한입니다. **생산자** | 작업 | Resource | 용도 | | --- | --- | --- | | DescribeStream, DescribeStreamSummary, DescribeStreamConsumer | Kinesis 데이터 스트림 | 레코드를 쓰려고 시도하기 전에 생산자는 스트림이 존재하고 활성 상태인지 여부, 샤드가 스트림에 포함되어 있는지 여부, 그리고 스트림에 소비자가 있는지 여부를 확인합니다. | | SubscribeToShard, RegisterStreamConsumer | Kinesis 데이터 스트림 | Kinesis 데이터 스트림 샤드에 소비자를 구독 및 등록합니다. | | PutRecord, PutRecords | Kinesis 데이터 스트림 | 레코드를 Kinesis Data Streams에 씁니다. | **소비자** | **작업** | **리소스** | **용도** | | --- | --- | --- | | DescribeStream | Kinesis 데이터 스트림 | 레코드를 읽으려고 시도하기 전에 소비자는 스트림이 존재하는지 및 활성 상태인지 여부와 샤드가 스트림에 포함되어 있는지 여부를 확인합니다. | | GetRecords, GetShardIterator | Kinesis 데이터 스트림 | Kinesis Data Streams 샤드에서 레코드를 읽습니다. | | CreateTable, DescribeTable, GetItem, PutItem, Scan, UpdateItem | Amazon DynamoDB 테이블 | Kinesis Client Library(KCL)를 사용하여 소비자를 개발하는 경우, 애플리케이션의 처리 상태를 추적하려면 DynamoDB 테이블에 대한 권한이 필요합니다. 시작된 첫 번째 소비자가 테이블을 생성합니다. | | DeleteItem | Amazon DynamoDB 테이블 | 소비자가 Kinesis Data Streams 샤드에서 분할/병합 작업을 수행하는 경우에 사용됩니다. | | PutMetricData | Amazon CloudWatch 로그 | 또한 KCL은 애플리케이션을 모니터링하는 데 유용한 지표를 CloudWatch에 업로드합니다. | 이 애플리케이션의 경우 위의 모든 권한을 부여하는 단일 IAM 정책을 생성합니다. 실제로 생산자에 대한 정책 하나와 소비자에 대한 정책 하나 등 정책 두 개의 생성을 고려해야 할 수 있습니다. **IAM 정책을 만들려면** 1. 새 스트림에 대한 ARN(Amazon 리소스 이름)을 찾습니다. **세부 정보** 탭 상단에 **스트림 ARN**으로 나열된 이 ARN을 찾을 수 있습니다. ARN 형식은 다음과 같습니다. ``` arn:aws:kinesis:region:account:stream/name ``` *리전* 리전 코드입니다(예: `us-west-2`). 자세한 내용은 [리전 및 가용 영역 개념](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html#concepts-regions-availability-zones)을 참조하십시오. *계정* AWS 계정 [설정에](https://console.aws.amazon.com/billing/home?#/account) 표시된 계정 ID입니다. *이름* [데이터 스트림 생성](tutorial-stock-data-kplkcl-create-stream.md)의 스트림 이름입니다. 이 경우 `StockTradeStream`입니다. 1. 소비자가 사용할 DynamoDB 테이블에 대한 ARN을 결정합니다(첫 번째 소비자 인스턴스에서 생성됨). 형식은 다음과 같아야 합니다. ``` arn:aws:dynamodb:region:account:table/name ``` 리전과 계정은 이전 단계와 동일한 위치에서 가져오지만 이때 *이름*은 소비자 애플리케이션에서 생성되고 사용되는 테이블의 이름입니다. 소비자가 사용하는 KCL은 애플리케이션 이름을 테이블 이름으로 사용합니다. 나중에 사용되는 애플리케이션 이름인 `StockTradesProcessor`를 사용합니다. 1. IAM 콘솔의 **정책**([https://console.aws.amazon.com/iam/home\$1policies](https://console.aws.amazon.com/iam/home#policies))에서 **정책 생성**을 선택합니다. IAM 정책을 사용한 첫 번째 작업인 경우 **시작하기**, **정책 생성**을 선택합니다. 1. **정책 생성기** 옆의 **선택**을 선택합니다. 1. **Amazon Kinesis**를 AWS 서비스로 선택합니다. 1. `DescribeStream`, `GetShardIterator`, `GetRecords`, `PutRecord` 및 `PutRecords`를 허용된 작업으로 선택합니다. 1. 1단계에서 생성한 ARN을 입력합니다. 1. 다음의 각각에 대해 **Add Statement(문 추가)**를 사용합니다. [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/ko_kr/streams/latest/dev/tutorial-stock-data-kplkcl-iam.html) 별표(`*`)는 ARN이 필요하지 않다고 지정할 때 사용됩니다. 이 경우에는 `PutMetricData` 작업이 간접적으로 호출된 CloudWatch에서 특정 리소스가 없기 때문입니다. 1. **다음 단계**를 선택합니다. 1. **Policy Name(정책 이름)**을 `StockTradeStreamPolicy`로 변경하고, 코드를 검토한 다음 **Create Policy(정책 생성)**를 선택합니다. 이를 통해 생성된 정책 문서는 다음과 같아야 합니다. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Sid": "Stmt123", "Effect": "Allow", "Action": [ "kinesis:DescribeStream", "kinesis:PutRecord", "kinesis:PutRecords", "kinesis:GetShardIterator", "kinesis:GetRecords", "kinesis:ListShards", "kinesis:DescribeStreamSummary", "kinesis:RegisterStreamConsumer" ], "Resource": [ "arn:aws:kinesis:us-west-2:111122223333:stream/StockTradeStream" ] }, { "Sid": "Stmt234", "Effect": "Allow", "Action": [ "kinesis:SubscribeToShard", "kinesis:DescribeStreamConsumer" ], "Resource": [ "arn:aws:kinesis:us-west-2:111122223333:stream/StockTradeStream/*" ] }, { "Sid": "Stmt456", "Effect": "Allow", "Action": [ "dynamodb:*" ], "Resource": [ "arn:aws:dynamodb:us-west-2:111122223333:table/StockTradesProcessor" ] }, { "Sid": "Stmt789", "Effect": "Allow", "Action": [ "cloudwatch:PutMetricData" ], "Resource": [ "*" ] } ] } ``` ------ **IAM 사용자 생성** 1. IAM 콘솔([https://console.aws.amazon.com/iam/](https://console.aws.amazon.com/iam/))을 엽니다. 1. **사용자** 페이지에서 **사용자 추가**를 선택합니다. 1. **사용자 이름**에 `StockTradeStreamUser`을 입력합니다. 1. **Access type(액세스 유형)**에서 **Programmatic access(프로그래밍 방식 액세스)**를 선택한 다음 **Next: Permissions(다음: 권한)**를 선택합니다. 1. **기존 정책 직접 첨부**를 선택합니다. 1. 생성한 정책의 이름으로 검색합니다. 정책 이름 왼쪽에 있는 확인란을 선택하고 **Next: Review(다음: 검코)**를 선택합니다. 1. 세부 정보와 요약을 검토하고 **Create user(사용자 생성)**를 선택합니다. 1. **Access key ID(액세스 키 ID)**를 복사하고 비공개로 저장합니다. **Secret access key(보안 액세스 키)**에서 **Show(표시)**를 선택하고 키도 비공개로 저장합니다. 1. 액세스 및 보안 키를 사용자만 액세스할 수 있는 안전한 위치에 있는 로컬 파일에 붙여넣습니다. 이 애플리케이션의 경우 ` ~/.aws/credentials`라는 파일 이름을 생성합니다(엄격한 권한 포함). 파일은 다음 형식이어야 합니다. ``` [default] aws_access_key_id=access key aws_secret_access_key=secret access key ``` **사용자에게 IAM 정책 연결** 1. IAM 콘솔에서 [정책](https://console.aws.amazon.com/iam/home?#policies)을 열고 **정책 작업**을 선택합니다. 1. `StockTradeStreamPolicy` 및 **Attach(연결)**를 선택합니다. 1. `StockTradeStreamUser` 및 **Attach Policy(정책 연결)**를 선택합니다. ## 다음 단계 [구현 코드 다운로드 및 빌드](tutorial-stock-data-kplkcl-download.md) # 구현 코드 다운로드 및 빌드 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)에 대한 스켈레톤 코드가 제공됩니다. 여기에는 주식 거래 스트림 수집(*생산자*)과 데이터 처리(*소비자*)를 위한 스텁 구현이 포함되어 있습니다. 다음 절차는 구현을 완료하는 방법을 보여줍니다. **구현 코드를 다운로드하고 빌드하려면** 1. [소스 코드](https://github.com/awslabs/amazon-kinesis-learning/tree/learning-module-1)를 컴퓨터에 다운로드합니다. 1. 소스 코드를 사용하여 선호하는 IDE에 프로젝트를 생성하면 제공된 디렉터리 구조를 따르게 됩니다. 1. 프로젝트에 다음 라이브러리를 추가합니다. + Amazon Kinesis Client Library(KCL) + AWS SDK + Apache HttpCore + Apache HttpClient + Apache Commons Lang + Apache Commons Logging + Guava(Google Core Libraries For Java) + Jackson Annotations + Jackson Core + Jackson Databind + Jackson Dataformat: CBOR + Joda Time 1. IDE에 따라 프로젝트가 자동으로 빌드될 수 있습니다. 그렇지 않으면 IDE에 적합한 단계를 사용하여 프로젝트를 빌드하십시오. 이러한 단계를 성공적으로 완료한 경우 이제 다음 단원([생산자 구현](tutorial-stock-data-kplkcl-producer.md))으로 이동할 준비가 되었습니다. 임의의 단계에서 빌드에 오류가 발생하는 경우 오류를 조사하고 수정한 다음 진행하십시오. ## 다음 단계 [[생산자 구현](tutorial-stock-data-kplkcl-producer.md)생산자 구현](tutorial-stock-data-kplkcl-producer.md) # 생산자 구현 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리[자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)](tutorial-stock-data-kplkcl.md)의 애플리케이션은 실제 주식 시장 거래 모니터링의 시나리오를 사용합니다. 다음 원칙은 이 시나리오를 생산자와 지원 코드 구조에 매핑하는 방법을 간략하게 설명합니다. 소스 코드를 참조하여 다음 정보를 검토하십시오. **StockTrade 클래스** 개별 주식 거래는 `StockTrade` 클래스의 인스턴스로 표시됩니다. 이 인스턴스에는 티커 기호, 가격, 공유 수, 거래 유형(구매 또는 판매), 거래를 고유하게 식별하는 ID 등의 속성이 포함됩니다. 이 클래스가 사용자를 위해 구현됩니다. **스트림 레코드** 스트림은 레코드의 시퀀스입니다. 레코드는 JSON 형식으로 된 `StockTrade` 인스턴스의 직렬화입니다. 예제: ``` { "tickerSymbol": "AMZN", "tradeType": "BUY", "price": 395.87, "quantity": 16, "id": 3567129045 } ``` **StockTradeGenerator 클래스** `StockTradeGenerator`에는 호출될 때마다 새로 생성된 임의의 주식 거래를 반환하는 `getRandomTrade()`라는 메서드가 있습니다. 이 클래스가 사용자를 위해 구현됩니다. **StockTradesWriter 클래스** 생산자의 `main` 메서드인 `StockTradesWriter`는 계속적으로 임의의 거래를 검색하고 다음 작업을 수행하여 Kinesis Data Streams에 전송합니다. 1. 스트림 이름과 리전 이름을 입력으로 읽습니다. 1. `AmazonKinesisClientBuilder`를 생성합니다. 1. 클라이언트 빌더를 사용하여 리전, 자격 증명 및 클라이언트 구성을 설정합니다. 1. 클라이언트 빌더를 사용하여 `AmazonKinesis` 클라이언트를 빌드합니다. 1. 스트림의 존재 여부와 활성 상태 여부를 확인합니다. 그렇지 않은 경우 오류로 종료됩니다. 1. 연속 루프에서 `StockTradeGenerator.getRandomTrade()` 메서드를 호출하고 `sendStockTrade` 메서드를 호출하여 100밀리초마다 거래를 스트림으로 전송합니다. `sendStockTrade` 클래스의 `StockTradesWriter` 메서드에는 다음 코드가 있습니다. ``` private static void sendStockTrade(StockTrade trade, AmazonKinesis kinesisClient, String streamName) { byte[] bytes = trade.toJsonAsBytes(); // The bytes could be null if there is an issue with the JSON serialization by the Jackson JSON library. if (bytes == null) { LOG.warn("Could not get JSON bytes for stock trade"); return; } LOG.info("Putting trade: " + trade.toString()); PutRecordRequest putRecord = new PutRecordRequest(); putRecord.setStreamName(streamName); // We use the ticker symbol as the partition key, explained in the Supplemental Information section below. putRecord.setPartitionKey(trade.getTickerSymbol()); putRecord.setData(ByteBuffer.wrap(bytes)); try { kinesisClient.putRecord(putRecord); } catch (AmazonClientException ex) { LOG.warn("Error sending record to Amazon Kinesis.", ex); } } ``` 다음 코드 세부 분석을 참조하십시오. + `PutRecord` API에는 바이트 어레이가 필요하며, `trade`를 JSON 형식으로 변환해야 합니다. 이 한 줄의 코드는 다음 작업을 수행합니다. ``` byte[] bytes = trade.toJsonAsBytes(); ``` + 거래를 전송하기 전에 새 `PutRecordRequest` 인스턴스(이 경우 `putRecord`라고 함)를 생성합니다. ``` PutRecordRequest putRecord = new PutRecordRequest(); ``` 각 `PutRecord` 호출에는 스트림 이름, 파티션 키 및 데이터 BLOB가 필요합니다. 다음 코드는 `putRecord` 메서드를 사용하여 `setXxxx()` 객체의 이러한 필드를 채웁니다. ``` putRecord.setStreamName(streamName); putRecord.setPartitionKey(trade.getTickerSymbol()); putRecord.setData(ByteBuffer.wrap(bytes)); ``` 이 예제는 특정 샤드에 레코드를 매핑하는 주식 티켓을 파티션 키로 사용합니다. 실제로 레코드가 스트림에 대해 균등하게 분산되도록 샤드당 수백 개 또는 수천 개의 파티션 키가 있어야 합니다. 스트림에 데이터를 추가하는 방법에 대한 자세한 내용은 [스트림에 데이터 추가](developing-producers-with-sdk.md#kinesis-using-sdk-java-add-data-to-stream) 단원을 참조하십시오. 이제 `putRecord`를 클라이언트에 전송할 준비가 되었습니다(`put` 작업). ``` kinesisClient.putRecord(putRecord); ``` + 오류 확인과 로깅 기능은 항상 유용한 추가 기능입니다. 이 코드는 오류 조건을 기록합니다. ``` if (bytes == null) { LOG.warn("Could not get JSON bytes for stock trade"); return; } ``` `put`넣기 작업에 try/catch 블록을 추가합니다. ``` try { kinesisClient.putRecord(putRecord); } catch (AmazonClientException ex) { LOG.warn("Error sending record to Amazon Kinesis.", ex); } ``` 이렇게 하는 이유는 네트워크 오류로 인해 또는 처리량 제한에 도달하여 병목 현상이 발생한 스트림으로 인해 Kinesis Data Streams `put` 작업이 실패할 수 있기 때문입니다. 데이터 손실을 방지하기 위해 재시도를 사용하는 것과 같이 `put` 작업에 대한 재시도 정책을 신중히 고려하는 것이 좋습니다. + 상태 로깅은 유용하지만 선택 사항입니다. ``` LOG.info("Putting trade: " + trade.toString()); ``` 여기에 표시된 생산자는 Kinesis Data Streams API 단일 레코드 기능인 `PutRecord`를 사용합니다. 실제로 개별 생산자가 많은 레코드를 생성하는 경우 `PutRecords`의 여러 레코드 기능을 사용하고 레코드의 배치를 한 번에 전송하는 것이 더 효율적인 경우가 많습니다. 자세한 내용은 [스트림에 데이터 추가](developing-producers-with-sdk.md#kinesis-using-sdk-java-add-data-to-stream) 단원을 참조하십시오. **생산자를 실행하려면** 1. 앞에서(IAM 사용자를 생성할 때) 검색한 액세스 키 및 보안 키 페어가 `~/.aws/credentials` 파일에 저장되었는지 확인합니다. 1. 다음과 같은 인수를 사용하여 `StockTradeWriter` 클래스를 실행합니다. ``` StockTradeStream us-west-2 ``` `us-west-2` 이외의 리전에 스트림을 생성한 경우 여기에 해당 리전을 대신 지정해야 합니다. 다음과 유사한 출력 화면이 표시되어야 합니다. ``` Feb 16, 2015 3:53:00 PM com.amazonaws.services.kinesis.samples.stocktrades.writer.StockTradesWriter sendStockTrade INFO: Putting trade: ID 8: SELL 996 shares of BUD for $124.18 Feb 16, 2015 3:53:00 PM com.amazonaws.services.kinesis.samples.stocktrades.writer.StockTradesWriter sendStockTrade INFO: Putting trade: ID 9: BUY 159 shares of GE for $20.85 Feb 16, 2015 3:53:01 PM com.amazonaws.services.kinesis.samples.stocktrades.writer.StockTradesWriter sendStockTrade INFO: Putting trade: ID 10: BUY 322 shares of WMT for $90.08 ``` 이제 주식 거래 스트림이 Kinesis Data Streams에서 수집됩니다. ## 다음 단계 [소비자 구현](tutorial-stock-data-kplkcl-consumer.md) # 소비자 구현 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리[자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)](tutorial-stock-data-kplkcl.md)의 소비자 애플리케이션은 [[생산자 구현](tutorial-stock-data-kplkcl-producer.md)생산자 구현](tutorial-stock-data-kplkcl-producer.md)에서 만든 주식 거래 스트림을 계속 처리합니다. 그런 다음 1분마다 매매된 가장 인기 있는 주식들을 출력합니다. 애플리케이션은 소비자 앱에 공통적인 과중한 업무를 많이 수행하는 Kinesis Client Library(KCL)를 기반으로 하여 빌드됩니다. 자세한 내용은 [KCL 1.x 소비자 개발](developing-consumers-with-kcl.md) 단원을 참조하십시오. 소스 코드를 참조하여 다음 정보를 검토하십시오. **StockTradesProcessor 클래스** 다음 작업을 수행하는 소비자의 기본 클래스가 제공됩니다. + 인수로 전달된 애플리케이션, 스트림 및 리전 이름을 읽습니다. + `~/.aws/credentials`에서 자격 증명을 읽습니다. + `RecordProcessorFactory` 인스턴스에 의해 구현된 `RecordProcessor`의 서버 인스턴스를 제공하는 `StockTradeRecordProcessor` 인스턴스를 생성합니다. + `RecordProcessorFactory` 인스턴스가 있는 KCL 워커와 스트림 이름, 보안 인증 및 애플리케이션 이름이 포함된 표준 구성을 생성합니다. + 워커는 각 샤드(이 소비자 인스턴스에 할당된 샤드)에 대해 새 스레드를 생성하며, 이 스레드는 Kinesis Data Streams에서 계속 반복적으로 레코드를 읽습니다. 그런 다음 `RecordProcessor` 인스턴스를 호출하여 수신한 각 일괄 레코드를 처리합니다. **StockTradeRecordProcessor 클래스** `RecordProcessor` 인스턴스를 구현하고 `initialize`, `processRecords` 및 `shutdown`의 세 가지 필수 메서드를 구현합니다. Kinesis Client Library는 `initialize` 및 `shutdown`을 사용하여 레코드 프로세서에 레코드 수신을 시작할 준비가 될 시점과 레코드 수신을 중지해야 할 시점을 각각 알리므로, 레코드 프로세스가 모든 애플리케이션별 설정 및 종료 작업을 수행할 수 있습니다. 이에 대한 코드가 제공됩니다. `processRecords` 메서드에서 기본 처리가 발생하며, 각 레코드에 대해 `processRecord`를 사용합니다. 이 후자의 메서드는 사용자에 대해 대부분의 빈 스켈레톤 코드로 제공되어 향후 설명할 다음 단계에서 구현됩니다. 원래 소스 코드에서 비어 있는 `processRecord`: `reportStats`, and `resetStats`에 대한 지원 메서드의 구현에 유의하십시오. `processRecords` 메서드가 구현되며 다음 단계를 수행합니다. + 전달된 각 레코드의 경우 `processRecord`를 호출합니다. + 마지막 보고 이후 1분 이상이 경과된 경우 최신 통계를 인쇄하는 `reportStats()`를 호출한 후 통계를 지우는 `resetStats()`를 호출하여 다음 간격에 새 레코드만 포함되도록 합니다. + 다음 보고 시간을 설정합니다. + 마지막 체크포인트 이후 1분 이상이 경과된 경우 `checkpoint()`를 호출합니다. + 다음 검사 시간을 설정합니다. 이 메서드는 보고 및 검사 속도에 대해 60초 간격을 사용합니다. 검사에 대한 자세한 내용은 [소비자에 대한 추가 정보](#tutorial-stock-data-kplkcl-consumer-supplement) 섹션을 참조하세요. **StockStats 클래스** 이 클래스는 시간에 따른 가장 인기 있는 주식에 대한 통계 추적 및 데이터 보존을 제공합니다. 다음 메서드가 포함된 이 코드가 제공됩니다. + `addStockTrade(StockTrade)`: 지정된 `StockTrade`를 실행 중인 통계에 삽입합니다. + `toString()`: 형식이 지정된 문자열로 통계를 반환합니다. 이 클래스는 각 주식에 대한 총 거래 수의 실행 개수와 최대 개수를 유지하여 가장 인기 있는 주식을 추적합니다. 그리고 주식 거래가 발생할 때마다 이러한 계수가 업데이트됩니다. 다음 단계에 표시된 대로 `StockTradeRecordProcessor` 클래스의 메서드에 코드를 추가합니다. **소비자를 구현하려면** 1. 정확한 크기의 `processRecord` 객체를 인스턴스화하고, 해당 객체에 레코드 데이터를 추가하고, 문제가 있는 경우 경고를 기록하여 `StockTrade` 메서드를 구현합니다. ``` StockTrade trade = StockTrade.fromJsonAsBytes(record.getData().array()); if (trade == null) { LOG.warn("Skipping record. Unable to parse record into StockTrade. Partition Key: " + record.getPartitionKey()); return; } stockStats.addStockTrade(trade); ``` 1. 간단한 `reportStats` 메서드를 구현합니다. 기본 설정에 대한 출력 형식을 자유롭게 수정합니다. ``` System.out.println("****** Shard " + kinesisShardId + " stats for last 1 minute ******\n" + stockStats + "\n" + "****************************************************************\n"); ``` 1. 마지막으로 `resetStats` 메서드를 구현합니다. 그러면 새 `stockStats` 인스턴스가 생성됩니다. ``` stockStats = new StockStats(); ``` **소비자를 실행하려면** 1. [[생산자 구현](tutorial-stock-data-kplkcl-producer.md)생산자 구현](tutorial-stock-data-kplkcl-producer.md)에서 작성한 생산자를 실행하여 시뮬레이션된 주식 거래 레코드를 스트림에 첨가합니다. 1. 앞에서(IAM 사용자를 생성할 때) 검색한 액세스 키 및 보안 키 페어가 `~/.aws/credentials` 파일에 저장되었는지 확인합니다. 1. 다음과 같은 인수를 사용하여 `StockTradesProcessor` 클래스를 실행합니다. ``` StockTradesProcessor StockTradeStream us-west-2 ``` `us-west-2` 이외의 리전에 스트림을 생성한 경우 여기에 해당 리전을 대신 지정해야 합니다. 1분 후 다음과 같은 출력이 표시되어야 하며, 그 이후로 매분마다 새로 고침됩니다. ``` ****** Shard shardId-000000000001 stats for last 1 minute ****** Most popular stock being bought: WMT, 27 buys. Most popular stock being sold: PTR, 14 sells. **************************************************************** ``` ## 소비자에 대한 추가 정보 [KCL 1.x 소비자 개발](developing-consumers-with-kcl.md) 및 다른 부분에서 설명하는 Kinesis Client Library의 이점을 잘 알고 있는 경우 여기에서 Kinesis Client Library를 사용해야 하는 이유에 대해 궁금할 수 있습니다. 단일 샤드 스트림과 단일 소비자 인스턴스만 사용하여 처리하는 경우에도 KCL을 사용하여 소비자를 구현하는 것이 훨씬 더 쉽습니다. 생산자 단원의 코드 구현 단계를 소비자와 비교하면 소비자 구현이 비교적 쉽다는 것을 알 수 있습니다. 이는 대부분 KCL이 제공하는 서비스로 인한 것입니다. 이 애플리케이션에서는 개별 레코드를 처리할 수 있는 레코드 프로세서 클래스를 구현하는 것에 중점을 둡니다. Kinesis Data Streams에서 레코드를 가져오는 방식에 대해서는 걱정할 필요가 없습니다. KCL은 새 레코드가 사용 가능할 때마다 레코드를 가져오고 레코드 프로세서를 간접적으로 호출합니다. 또한 얼마나 많은 샤드와 소비자 인스턴스가 있는지에 대해서도 걱정할 필요가 없습니다. 스트림이 확장되면 둘 이상의 샤드 또는 소비자 인스턴스를 처리하기 위해 애플리케이션을 다시 작성할 필요가 없습니다. **체크포인트 수행이라는 용어는 지금까지 사용 및 처리된 데이터 레코드까지 스트림의 지점을 기록하는 것을 의미합니다. 애플리케이션이 충돌하면 스트림의 시작 부분이 아닌 해당 지점부터 스트림을 읽습니다. 검사 주체와 다양한 디자인 패턴 및 이에 대한 모범 사례는 이 장의 범위를 벗어나지만, 프로덕션 환경에서 직면할 수 있는 사항입니다. [[생산자 구현](tutorial-stock-data-kplkcl-producer.md)생산자 구현](tutorial-stock-data-kplkcl-producer.md)에서 배운 바와 같이, Kinesis Data Streams API의 `put` 작업은 **파티션 키를 입력으로 사용합니다. Kinesis Data Streams는 파티션 키를 메커니즘으로 사용하여 레코드를 여러 샤드로 분할합니다(스트림에 샤드가 여러 개 있는 경우). 동일한 파티션 키는 항상 동일한 샤드에 라우팅됩니다. 이를 통해 특정 샤드를 처리하는 소비자는 동일한 파티션 키가 있는 레코드는 해당 소비자에게만 전송되며, 다른 소비자에 전송될 수 없다는 가정에 기반하여 설계할 수 있습니다. 따라서 소비자의 작업자는 필요한 데이터가 누락될 수 있다는 걱정 없이 동일한 파티션 키가 있는 모든 레코드를 집계할 수 있습니다. 이 애플리케이션에서 소비자의 레코드 처리는 집약적이지 않으므로 샤드 하나를 사용하고 KCL 스레드와 동일한 스레드에서 처리할 수 있습니다. 하지만 실제로 먼저 샤드 수를 확장하는 것을 고려해 보십시오. 일부 경우에는 처리를 다른 스레드로 전환하거나, 레코드 처리가 집약적으로 예상될 경우 스레드 풀을 사용할 수 있습니다. 이러한 방식으로 KCL은 다른 스레드가 레코드를 병렬로 처리하는 동안 새 레코드를 더욱 신속하게 가져올 수 있습니다. 멀티스레드 디자인은 중요한 요소이며 고급 기술을 사용하여 접근해야 합니다. 따라서 샤드 수를 늘리는 것은 일반적으로 확장하기 위한 가장 효과적인 방식입니다. ## 다음 단계 [(선택 사항) 소비자 확장](tutorial-stock-data-kplkcl-consumer-extension.md) # (선택 사항) 소비자 확장 [자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리[자습서: KPL 및 KCL 1.x를 사용하여 실시간 주식 데이터 처리](tutorial-stock-data-kplkcl.md)](tutorial-stock-data-kplkcl.md)의 애플리케이션은 이미 해당 용도에 충분할 수 있습니다. 이 단원은 선택 사항이며 더 자세한 시나리오를 위해 소비자 코드를 확장할 수 있는 방법을 보여줍니다. 1분마다 가장 큰 판매 주문을 파악하려는 경우, 세 위치에서 `StockStats` 클래스를 수정하여 이 새로운 우선순위를 수용할 수 있습니다. **소비자를 확장하려면** 1. 새 인스턴스 변수를 추가합니다. ``` // Ticker symbol of the stock that had the largest quantity of shares sold private String largestSellOrderStock; // Quantity of shares for the largest sell order trade private long largestSellOrderQuantity; ``` 1. 다음 코드를 `addStockTrade`에 추가합니다. ``` if (type == TradeType.SELL) { if (largestSellOrderStock == null || trade.getQuantity() > largestSellOrderQuantity) { largestSellOrderStock = trade.getTickerSymbol(); largestSellOrderQuantity = trade.getQuantity(); } } ``` 1. `toString` 메서드를 수정하여 추가 정보를 인쇄합니다. ``` public String toString() { return String.format( "Most popular stock being bought: %s, %d buys.%n" + "Most popular stock being sold: %s, %d sells.%n" + "Largest sell order: %d shares of %s.", getMostPopularStock(TradeType.BUY), getMostPopularStockCount(TradeType.BUY), getMostPopularStock(TradeType.SELL), getMostPopularStockCount(TradeType.SELL), largestSellOrderQuantity, largestSellOrderStock); } ``` 이제 소비자를 실행하면(생산자도 실행함을 주의) 다음과 비슷한 출력 화면이 표시되어야 합니다. ``` ****** Shard shardId-000000000001 stats for last 1 minute ****** Most popular stock being bought: WMT, 27 buys. Most popular stock being sold: PTR, 14 sells. Largest sell order: 996 shares of BUD. **************************************************************** ``` ## 다음 단계 [리소스 정리](tutorial-stock-data-kplkcl-finish.md) # 리소스 정리 Kinesis 데이터 스트림 사용 요금을 결제하고 있으므로 작업이 완료되면 이 스트림과 해당 Amazon DynamoDB 테이블을 삭제해야 합니다. 레코드를 전송하거나 가져오지 않는 경우에도 활성 스트림에 대해 일반 요금이 부과됩니다. 이는 활성 스트림이 레코드를 가져오라는 요청과 들어오는 레코드를 지속적으로 "수신"하여 리소스를 사용하고 있기 때문입니다. **스트림과 테이블을 삭제하려면** 1. 계속 실행 중인 생산자와 소비자를 종료합니다. 1. [https://console.aws.amazon.com/kinesis](https://console.aws.amazon.com/kinesis)에서 Kinesis 콘솔을 엽니다. 1. 이 애플리케이션에 대해 생성한 스트림을 선택합니다(`StockTradeStream`). 1. **Delete Stream(스트림 삭제)**을 선택합니다. 1. [https://console.aws.amazon.com/dynamodb/](https://console.aws.amazon.com/dynamodb/)에서 DynamoDB 콘솔을 엽니다. 1. `StockTradesProcessor` 테이블을 삭제합니다. ## 요약 대량의 데이터를 거의 실시간으로 처리하는 데에는 복잡한 코드를 작성하거나 거대한 인프라를 개발할 필요가 없습니다. 기본 로직을 작성하여 소량의 데이터를 처리할 수 있지만(예: `processRecord(Record)` 작성), Kinesis Data Streams를 사용하여 확장할 수 있으므로 대량의 스트리밍 데이터에도 효과적입니다. Kinesis Data Streams에서 자동으로 처리되므로 처리를 확장하는 방법에 대해 걱정할 필요가 없습니다. 사용자는 스트리밍 레코드를 Kinesis Data Streams에 전송하고 수신된 각 새 레코드를 처리하는 로직을 작성하면 됩니다. 이 애플리케이션에 대한 몇 가지 잠재적 개선 사항이 있습니다. **모든 샤드에 대한 집계** 현재는 단일 샤드에서 단일 작업자가 수신한 데이터 레코드의 집계로 인해 생성된 통계를 가져옵니다. (단일 애플리케이션에서 동시에 둘 이상의 작업자가 하나의 샤드를 처리할 수 없음) 물론, 샤드를 확장하여 샤드가 두 개 이상인 경우 모든 샤드에 대해 집계할 수 있습니다. 각 작업자의 출력이 단일 샤드가 있는 다른 스트림으로 공급되어 첫 단계의 출력을 집계하는 작업자가 처리하는 파이프라인 아키텍처를 구축하여 이를 수행할 수 있습니다. 첫 단계의 데이터가 제한(샤드마다 분당 샘플 하나)되므로 샤드 하나로 쉽게 처리할 수 있습니다. **처리 확장** 스트림이 여러 샤드가 있도록 확장되면(여러 생산자가 데이터를 전송하기 때문) 더 많은 작업자를 추가하는 방식으로 처리가 확장됩니다. Amazon EC2 인스턴스에서 워커를 실행하고 오토 스케일링을 사용할 수 있습니다. **Amazon S3/DynamoDB/Amazon Redshift/Storm에 대한 커넥터 사용** 스트림이 지속적으로 처리되면 출력을 다른 대상으로 전송할 수 있습니다.는 Kinesis Data Streams를 다른 AWS 서비스 및 타사 도구와 통합하기 위한 [커넥터를](https://github.com/awslabs/amazon-kinesis-connectors) AWS 제공합니다. ## 다음 단계 + Kinesis Data Streams API 작업 사용에 대한 자세한 내용은 [에서 Amazon Kinesis Data Streams API를 사용하여 생산자 개발 AWS SDK for Java](developing-producers-with-sdk.md), [를 사용하여 공유 처리량 소비자 개발 AWS SDK for Java](developing-consumers-with-sdk.md) 및 [Kinesis 데이터 스트림 생성 및 관리](working-with-streams.md) 섹션을 참조하세요. + Kinesis Client Library에 대한 자세한 내용은 [KCL 1.x 소비자 개발](developing-consumers-with-kcl.md) 섹션을 참조하세요. + 애플리케이션을 최적화하는 방법에 대한 자세한 내용은 [Amazon Kinesis Data Streams 소비자 최적화Kinesis Data Streams 소비자 최적화](advanced-consumers.md) 단원을 참조하십시오.