자습서: API Gateway에서 REST API를 Amazon Kinesis 프록시로 생성 - Amazon API Gateway
API에서 Kinesis 액세스를 위한 IAM 역할 및 정책 생성API를 Kinesis 프록시로 생성하기 시작Kinesis에 스트림 나열Kinesis에서 스트림 생성, 설명, 삭제Kinesis의 스트림에서 레코드 가져오기 및 추가

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

자습서: API Gateway에서 REST API를 Amazon Kinesis 프록시로 생성

이 페이지에서는 AWS 유형의 통합을 통해 REST API를 생성하고 구성하여 Kinesis에 액세스하는 방법을 설명합니다.

참고

API Gateway API를 Kinesis와 통합하려면 API Gateway와 Kinesis 서비스를 모두 사용할 수 있는 리전을 선택해야 합니다. 리전 사용 가능성은 서비스 엔드포인트 및 할당량 단원을 참조하십시오.

이 예시에서는 클라이언트가 다음 작업을 수행하도록 허용하는 예제 API를 생성합니다.

  1. Kinesis에서 사용자의 사용 가능한 스트림 나열

  2. 지정된 스트림 생성, 설명 또는 삭제

  3. 지정된 스트림에서 데이터 레코드 읽기 또는 쓰기

위의 작업을 수행하기 위해 API에서는 다양한 리소스에 대해 각각 다음을 호출하는 메서드를 노출합니다.

  1. Kinesis의 ListStreams 작업

  2. CreateStream, DescribeStream 또는 DeleteStream 작업

  3. Kinesis의 GetRecords 또는 PutRecords(PutRecord 포함) 작업

특히, API를 다음과 같이 빌드합니다.

  • API /streams 리소스에 HTTP GET 메서드를 노출하고 이 메서드를 Kinesis의 ListStreams작업과 통합하여 호출자 계정의 스트림을 나열합니다.

  • API /streams/{stream-name} 리소스에 HTTP POST 메서드를 노출하고 이 메서드를 Kinesis의 CreateStream작업과 통합하여 호출자 계정에 명명된 스트림을 생성합니다.

  • API /streams/{stream-name} 리소스에 HTTP GET 메서드를 노출하고 이 메서드를 Kinesis의 DescribeStream작업과 통합하여 호출자 계정의 명명된 스트림을 설명합니다.

  • API /streams/{stream-name} 리소스에 HTTP DELETE 메서드를 노출하고 이 메서드를 Kinesis의 DeleteStream작업과 통합하여 호출자 계정에서 스트림을 삭제합니다.

  • API /streams/{stream-name}/record 리소스에 HTTP PUT 메서드를 노출하고 이 메서드를 Kinesis의 PutRecord작업과 통합합니다. 그러면 클라이언트가 단일 데이터 레코드를 명명된 스트림에 추가할 수 있습니다.

  • API /streams/{stream-name}/records 리소스에 HTTP PUT 메서드를 노출하고 이 메서드를 Kinesis의 PutRecords작업과 통합합니다. 그러면 클라이언트가 데이터 레코드 목록을 명명된 스트림에 추가할 수 있습니다.

  • API /streams/{stream-name}/records 리소스에 HTTP GET 메서드를 노출하고 이 메서드를 Kinesis의 GetRecords작업과 통합합니다. 그러면 클라이언트가 지정된 샤드 반복기를 사용하여 명명된 스트림에 데이터 레코드를 나열할 수 있습니다. 샤드 반복기는 순차적으로 데이터 레코드 읽기를 시작할 샤드 위치를 지정합니다.

  • API /streams/{stream-name}/sharditerator 리소스에 HTTP GET 메서드를 노출하고 이 메서드를 Kinesis의 GetShardIterator작업과 통합합니다. 이 헬퍼 메서드는 Kinesis의 ListStreams 작업에 제공되어야 합니다.

여기에 제공된 지침을 다른 Kinesis 작업에 적용할 수 있습니다. Kinesis 작업의 전체 목록은 Amazon Kinesis API 참조를 참조하십시오.

API Gateway 콘솔을 사용하여 샘플 API를 생성하는 대신 API Gateway API 가져오기를 사용하여 샘플 API를 API Gateway로 가져올 수 있습니다. API 가져오기를 사용하는 방법에 대한 자세한 내용은 OpenAPI를 사용하여 REST API 구성 단원을 참조하세요.

API에서 Kinesis 액세스를 위한 IAM 역할 및 정책 생성

API에서 Kinesis 작업을 간접적으로 호출하도록 허용하려면 적절한 IAM 정책을 IAM 역할에 연결해야 합니다.

AWS 서비스 프록시 실행 역할을 만들려면

  1. 에 AWS Management Console 로그인하고 https://console.aws.amazon.com/iam/ 에서 IAM 콘솔을 엽니다.

  2. [Roles]를 선택합니다.

  3. 역할 생성을 선택합니다.

  4. 신뢰할 수 있는 개체 유형 선택에서 AWS 서비스를 선택한 다음 API Gateway를 선택하고 API Gateway에서 로그를 로그로 푸시할 수 있도록 허용을 CloudWatch 선택합니다.

  5. 다음을 선택한 후 다시 다음을 선택합니다.

  6. 역할 이름에 APIGatewayKinesisProxyPolicy를 입력한 다음 역할 생성을 선택합니다.

  7. 역할 목록에서 방금 생성한 역할을 선택합니다. 스크롤하거나 검색 창을 사용하여 역할을 찾을 수 있습니다.

  8. 선택한 역할에 대해 권한 추가 탭을 선택합니다.

  9. 드롭다운 목록에서 정책 연결을 선택합니다.

  10. 검색 창에 AmazonKinesisFullAccess를 입력하고 권한 추가를 선택합니다.

    참고

    이 자습서에서는 단순화를 위해 관리형 정책을 사용합니다. 가장 좋은 방법은 필요한 최소 권한을 부여하는 자체 IAM 정책을 생성하는 것입니다.

  11. 새로 생성된 역할 ARN은 나중에 사용할 수 있도록 기록해 둡니다.

API를 Kinesis 프록시로 생성하기 시작

다음 단계에 따라 API Gateway 콘솔에서 API를 생성합니다.

Kinesis용 AWS 서비스 프록시로 API를 생성하려면

  1. https://console.aws.amazon.com/apigateway에서 API Gateway 콘솔에 로그인합니다.

  2. API Gateway를 처음 사용하는 경우, 서비스의 기능을 소개하는 페이지가 나타납니다. REST API에서 빌드를 선택합니다. 예제 API 생성 팝업이 나타나면 확인을 선택합니다.

    API Gateway를 처음 사용하는 것이 아닌 경우 API 생성을 선택합니다. REST API에서 빌드를 선택합니다.

  3. 새 API(New API)를 선택합니다.

  4. API 이름에 KinesisProxy를 입력합니다. 다른 모든 필드에 대한 기본값을 그대로 유지합니다.

  5. (선택 사항) 설명에 설명을 입력합니다.

  6. API 생성(Create API)을 선택합니다.

API가 생성되면 API Gateway 콘솔에 리소스(Resources) 페이지가 표시됩니다. 이 페이지에는 API의 루트(/) 리소스만 포함되어 있습니다.

Kinesis에 스트림 나열

Kinesis는 다음의 REST API 호출을 통해 ListStreams 작업을 지원합니다.

POST /?Action=ListStreams HTTP/1.1
Host: kinesis.<region>.<domain>
Content-Length: <PayloadSizeBytes>
User-Agent: <UserAgentString>
Content-Type: application/x-amz-json-1.1
Authorization: <AuthParams>
X-Amz-Date: <Date>
        
{
   ...
}

위의 REST API 요청에서 작업은 Action 쿼리 파라미터에 지정합니다. 또는 다음과 같이 X-Amz-Target 헤더에서 작업을 지정할 수도 있습니다.

POST / HTTP/1.1
Host: kinesis.<region>.<domain>
Content-Length: <PayloadSizeBytes>
User-Agent: <UserAgentString>
Content-Type: application/x-amz-json-1.1
Authorization: <AuthParams>
X-Amz-Date: <Date>
X-Amz-Target: Kinesis_20131202.ListStreams        
{
   ...
}

이 자습서에서는 쿼리 파라미터를 사용하여 작업을 지정합니다.

API에 Kinesis 작업을 표시하려면 /streams 리소스를 API의 루트에 추가합니다. 그런 다음 리소스에 GET 메서드를 설정하고 그 메서드를 Kinesis의 ListStreams 작업과 통합합니다.

다음 절차에서는 API Gateway 콘솔을 사용하여 Kinesis 스트림을 나열하는 방법에 대해 설명합니다.

API Gateway 콘솔을 사용하여 Kinesis 스트림을 나열하려면

  1. / 리소스를 선택한 다음 리소스 생성을 선택합니다.

  2. 리소스 이름에 streams을 입력합니다.

  3. 오리진 간 리소스 공유(CORS)를 꺼진 상태로 둡니다.

  4. 리소스 생성을 선택합니다.

  5. /streams 리소스를 선택한 후 메서드 생성을 선택하고 다음을 수행합니다.

    1. 메서드 유형에서 GET을 선택합니다.

      참고

      클라이언트에 의해 호출된 메서드에 대한 HTTP 동사는 백엔드에서 요구하는 통합에 대한 HTTP 동사와 다를 수 있습니다. 여기에서는 GET을 선택했는데, 목록 스트림이 본질적으로 읽기(READ) 작업이기 때문입니다.

    2. 통합 유형에서 AWS 서비스를 선택합니다.

    3. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

    4. AWS 서비스에서 Kinesis를 선택합니다.

    5. AWS 하위 도메인은 비워 둡니다.

    6. HTTP 메서드(HTTP method)에 대해 POST를 선택합니다.

      참고

      여기에서는 POST를 선택했는데, Kinesis의 경우 ListStreams 작업이 POST로 호출되기 때문입니다.

    7. 작업 유형에서 사용자 작업 이름을 선택합니다.

    8. 함수 이름에 ListStreams를 입력합니다.

    9. 실행 역할에 실행 역할의 ARN을 입력합니다.

    10. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

    11. 메서드 생성을 선택합니다.

  6. 통합 요청 탭의 통합 요청 설정에서 편집을 선택합니다.

  7. 요청 본문 패스스루에서 정의된 템플릿이 없는 경우(권장)를 선택합니다.

  8. URL 요청 헤더 파라미터를 선택하고 다음을 수행합니다.

    1. 요청 헤더 파라미터 추가를 선택합니다.

    2. 이름에 Content-Type를 입력합니다.

    3. 다음에서 매핑됨에 'application/x-amz-json-1.1'을 입력합니다.

    Content-Type 헤더를 'application/x-amz-json-1.1'의 정적 값으로 설정하는 요청 파라미터 매핑을 사용하여 입력이 특정 버전의 JSON임을 Kinesis에 알립니다.

  9. 매핑 템플릿을 선택한 다음 매핑 템플릿 추가를 선택하고 다음을 수행합니다.

    1. 콘텐츠 유형에 application/json을 입력합니다.

    2. 템플릿 본문에 {}를 입력합니다.

    3. 저장을 선택합니다.

    ListStreams요청은 다음 JSON 형식의 페이로드를 사용합니다.

    
{
    "ExclusiveStartStreamName": "string",
    "Limit": number
}

    하지만 속성은 선택 사항입니다. 여기에서는 기본값을 사용하기 위해 비어 있는 JSON 페이로드를 선택했습니다.

  10. Kinesis에서 ListStreams 작업을 호출하는 GET 메서드를 /streams 리소스에서 테스트합니다.

    테스트 탭을 선택합니다. 탭을 표시하려면 오른쪽 화살표 버튼을 선택해야 할 수도 있습니다.

    테스트를 선택하여 메서드를 테스트합니다.

    Kinesis에서 이미 두 개의 스트림("myStream" 및 "yourStream")을 생성했다면, 테스트가 다음의 페이로드가 포함된 200 OK 응답을 반환할 것입니다.

    
{
     "HasMoreStreams": false,
     "StreamNames": [
         "myStream",
         "yourStream"
     ]
}

Kinesis에서 스트림 생성, 설명, 삭제

Kinesis에서 스트림을 생성, 설명, 삭제하는 동안 각각 다음 Kinesis REST API 요청을 생성합니다.


POST /?Action=CreateStream HTTP/1.1
Host: kinesis.region.domain
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "ShardCount": number,
    "StreamName": "string"
}                               
            

POST /?Action=DescribeStream HTTP/1.1
Host: kinesis.region.domain
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "StreamName": "string"
}                                          
            

POST /?Action=DeleteStream HTTP/1.1
Host: kinesis.region.domain
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "StreamName":"string"
}

필요한 입력을 메서드 요청의 JSON 페이로드로 수락하고 페이로드를 통합 요청으로 전달하는 API를 빌드할 수 있습니다. 하지만 메서드와 통합 요청 사이와 메서드와 통합 응답 사이에 더 많은 데이터 매핑 예제를 제공하기 위해 API를 약간 다르게 생성합니다.

리소스에GET,POST, Delete HTTP 메서드를 노출합니다. to-be-named Stream {stream-name} 경로 변수를 스트림 리소스의 자리 표시자로 사용하고 이들 API 메서드를 각각 Kinesis의 DescribeStream, CreateStream 및 DeleteStream 작업과 통합합니다. 클라이언트에서 다른 입력 데이터를 헤더, 쿼리 파라미터 또는 메서드 요청의 페이로드로 전달하도록 해야 합니다, 데이터를 필요한 통합 요청 페이로드로 변환하는 매핑 템플릿을 제공합니다.

{stream-name} 리소스를 생성하려면

  1. /stream 리소스를 선택한 다음 리소스 생성을 선택합니다.

  2. 프록시 리소스는 꺼진 상태로 둡니다.

  3. 리소스 경로에서 /streams를 선택합니다.

  4. 리소스 이름에 {stream-name}을 입력합니다.

  5. 오리진 간 리소스 공유(CORS)를 꺼진 상태로 둡니다.

  6. 리소스 생성을 선택합니다.

스트림 리소스에서 GET 메서드를 구성하고 테스트하는 방법

  1. /{stream-name} 리소스를 선택한 다음 [메소드 생성] 을 선택합니다.

  2. 메서드 유형에서 GET을 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 DescribeStream를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

  13. 통합 요청 섹션에서 다음 URL 요청 헤더 파라미터를 추가합니다.

    Content-Type: 'x-amz-json-1.1'

    작업 절차는 GET /streams 메서드에 대한 요청 파라미터 매핑을 설정하는 절차와 같습니다.

  14. 다음의 본문 매핑 템플릿을 추가하여 데이터를 GET /streams/{stream-name} 메서드 요청에서 POST /?Action=DescribeStream 통합 요청으로 매핑합니다.

    {
    "StreamName": "$input.params('stream-name')"
}

    이 매핑 템플릿은 메서드 요청의 DescribeStream 경로 파라미터 값으로부터 Kinesis의 stream-name 작업에 대한 필수 통합 요청 페이로드를 생성합니다.

  15. Kinesis에서 DescribeStream 작업을 간접적으로 호출하는 GET /stream/{stream-name} 메서드를 테스트하려면 테스트 탭을 선택합니다.

  16. 경로의 stream-name에 기존 Kinesis 스트림의 이름을 입력합니다.

  17. 테스트를 선택합니다. 테스트가 성공하면 다음과 비슷한 페이로드와 함께 200 OK 응답이 반환됩니다.

    {
  "StreamDescription": {
    "HasMoreShards": false,
    "RetentionPeriodHours": 24,
    "Shards": [
      {
        "HashKeyRange": {
          "EndingHashKey": "68056473384187692692674921486353642290",
          "StartingHashKey": "0"
        },
        "SequenceNumberRange": {
          "StartingSequenceNumber": "49559266461454070523309915164834022007924120923395850242"
        },
        "ShardId": "shardId-000000000000"
      },
      ...
      {
        "HashKeyRange": {
          "EndingHashKey": "340282366920938463463374607431768211455",
          "StartingHashKey": "272225893536750770770699685945414569164"
        },
        "SequenceNumberRange": {
          "StartingSequenceNumber": "49559266461543273504104037657400164881014714369419771970"
        },
        "ShardId": "shardId-000000000004"
      }
    ],
    "StreamARN": "arn:aws:kinesis:us-east-1:12345678901:stream/myStream",
    "StreamName": "myStream",
    "StreamStatus": "ACTIVE"
  }
}

    API를 배포한 후에 이 API 메서드에 대해 REST 요청을 할 수 있습니다.

    
GET https://your-api-id.execute-api.region.amazonaws.com/stage/streams/myStream HTTP/1.1
Host: your-api-id.execute-api.region.amazonaws.com
Content-Type: application/json
Authorization: ...
X-Amz-Date: 20160323T194451Z
스트림 리소스에서 POST 메서드를 구성하고 테스트하는 방법

  1. /{stream-name} 리소스를 선택한 다음 [메소드 생성] 을 선택합니다.

  2. 메서드 유형에서 POST를 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 CreateStream를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

  13. 통합 요청 섹션에서 다음 URL 요청 헤더 파라미터를 추가합니다.

    Content-Type: 'x-amz-json-1.1'

    작업 절차는 GET /streams 메서드에 대한 요청 파라미터 매핑을 설정하는 절차와 같습니다.

  14. 다음의 본문 매핑 템플릿을 추가하여 데이터를 POST /streams/{stream-name} 메서드 요청에서 POST /?Action=CreateStream 통합 요청으로 매핑합니다.

    {
    "ShardCount": #if($input.path('$.ShardCount') == '') 5 #else $input.path('$.ShardCount') #end,
    "StreamName": "$input.params('stream-name')"
}

    앞서 다룬 매핑 템플릿에서, 클라이언트가 메서드 요청 페이로드에서 값을 지정하지 않은 경우 ShardCount를 고정값 5로 설정합니다.

  15. Kinesis에서 CreateStream 작업을 간접적으로 호출하는 POST /stream/{stream-name} 메서드를 테스트하려면 테스트 탭을 선택합니다.

  16. 경로의 stream-name에 새 Kinesis 스트림의 이름을 입력합니다.

  17. 테스트를 선택합니다. 테스트가 성공하면 아무 데이터 없이 200 OK 응답이 반환됩니다.

    API를 배포한 후에 스트림 리소스 상에서 POST 메서드에 대한 REST API 요청을 만들어 Kinesis에서 CreateStream 작업을 호출할 수도 있습니다.

    
POST https://your-api-id.execute-api.region.amazonaws.com/stage/streams/yourStream HTTP/1.1
Host: your-api-id.execute-api.region.amazonaws.com
Content-Type: application/json
Authorization: ...
X-Amz-Date: 20160323T194451Z

{ 
    "ShardCount": 5
}
스트림 리소스에서 DELETE 메서드를 구성하고 테스트하는 방법

  1. /{stream-name} 리소스를 선택한 다음 [메소드 생성] 을 선택합니다.

  2. 메서드 유형에서 DELETE를 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 DeleteStream를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

  13. 통합 요청 섹션에서 다음 URL 요청 헤더 파라미터를 추가합니다.

    Content-Type: 'x-amz-json-1.1'

    작업 절차는 GET /streams 메서드에 대한 요청 파라미터 매핑을 설정하는 절차와 같습니다.

  14. 다음의 본문 매핑 템플릿을 추가하여 데이터를 DELETE /streams/{stream-name} 메서드 요청에서 POST /?Action=DeleteStream의 해당 통합 요청으로 매핑합니다.

    {
    "StreamName": "$input.params('stream-name')"
}

    이 매핑 템플릿은 클라이언트에서 제공한 DELETE /streams/{stream-name}의 URL 경로 이름으로부터 stream-name 작업에 대한 필수 입력을 생성합니다.

  15. Kinesis에서 DeleteStream 작업을 간접적으로 호출하는 DELETE /stream/{stream-name} 메서드를 테스트하려면 테스트 탭을 선택합니다.

  16. 경로의 stream-name에 기존 Kinesis 스트림의 이름을 입력합니다.

  17. 테스트를 선택합니다. 테스트가 성공하면 아무 데이터 없이 200 OK 응답이 반환됩니다.

    API를 배포한 후에 스트림 리소스 상에서 다음과 같은 DELETE 메서드에 대한 REST API 요청을 만들어 Kinesis에서 DeleteStream 작업을 호출할 수도 있습니다.

    
DELETE https://your-api-id.execute-api.region.amazonaws.com/stage/streams/yourStream HTTP/1.1
Host: your-api-id.execute-api.region.amazonaws.com
Content-Type: application/json
Authorization: ...
X-Amz-Date: 20160323T194451Z

{}

Kinesis의 스트림에서 레코드 가져오기 및 추가

Kinesis에서 스트림을 생성한 후 데이터 레코드를 스트림에 추가하고 스트림에서 데이터를 읽을 수 있습니다. 데이터 레코드를 추가하려면 Kinesis에서 PutRecords또는 PutRecord작업을 호출해야 합니다. 첫 번째 작업은 여러 레코드를 추가하고 두 번째 작업은 단일 레코드를 스트림에 추가합니다.


POST /?Action=PutRecords HTTP/1.1
Host: kinesis.region.domain
Authorization: AWS4-HMAC-SHA256 Credential=..., ...
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "Records": [
        {
            "Data": blob,
            "ExplicitHashKey": "string",
            "PartitionKey": "string"
        }
    ],
    "StreamName": "string"
}

또는


POST /?Action=PutRecord HTTP/1.1
Host: kinesis.region.domain
Authorization: AWS4-HMAC-SHA256 Credential=..., ...
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "Data": blob,
    "ExplicitHashKey": "string",
    "PartitionKey": "string",
    "SequenceNumberForOrdering": "string",
    "StreamName": "string"
}

여기서 StreamName은 레코드를 추가할 타겟 스트림을 식별합니다. StreamName, Data 및 PartitionKey는 필요한 입력 데이터입니다. 이 예에서는 모든 선택적 입력 데이터에 대해 기본값을 사용하며 메서드 요청에 대한 입력에서 해당 값을 명시적으로 지정하지 않습니다.

Kinesis에서 데이터를 읽는 것은 작업을 호출하는 것과 같습니다. GetRecords 

POST /?Action=GetRecords HTTP/1.1
Host: kinesis.region.domain
Authorization: AWS4-HMAC-SHA256 Credential=..., ...
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes

{
    "ShardIterator": "string",
    "Limit": number
}

샤드 반복기를 가져오기 위한 다음 Kinesis 작업에 표시된 대로 레코드를 가져올 소스 스트림은 필수 ShardIterator 값으로 지정됩니다.


POST /?Action=GetShardIterator HTTP/1.1
Host: kinesis.region.domain
Authorization: AWS4-HMAC-SHA256 Credential=..., ...
...
Content-Type: application/x-amz-json-1.1
Content-Length: PayloadSizeBytes
                
{
    "ShardId": "string",
    "ShardIteratorType": "string",
    "StartingSequenceNumber": "string",
    "StreamName": "string"
}

GetRecords 및 PutRecords 작업의 경우 명명된 스트림 리소스(GET)에 추가되는 PUT 리소스에 각각 /records 및 /{stream-name} 메서드를 노출합니다. 마찬가지로 PutRecord 리소스에 PUT 작업을 /record 메서드로 노출합니다.

GetRecords 작업은 ShardIterator 헬퍼 작업을 호출하여 GetShardIterator 값을 입력으로 가져오므로 GET 리소스(ShardIterator)에 /sharditerator 헬퍼 메서드를 노출합니다.

/record, /records 및 /sharditerator 리소스를 생성하려면

  1. /{stream-name} 리소스를 선택한 다음 리소스 생성을 선택합니다.

  2. 프록시 리소스는 꺼진 상태로 둡니다.

  3. 리소스 경로에서 /{stream-name}를 선택합니다.

  4. 리소스 이름에 record을 입력합니다.

  5. 오리진 간 리소스 공유(CORS)를 꺼진 상태로 둡니다.

  6. 리소스 생성을 선택합니다.

  7. 이전 단계를 반복하여 /records와 /sharditerator 리소스를 생성합니다. 최종 API는 다음과 같아야 합니다.

    API에 대한 Records:GET|PUT|PUT|GET 메서드 생성

다음 네 가지 절차에서는 각 메서드를 설정하는 방법, 메서드 요청에서 통합 요청으로 데이터를 매핑하는 방법, 메서드를 테스트하는 방법을 설명합니다.

Kinesis에서 PutRecord를 호출할 수 있도록 PUT /streams/{stream-name}/record 메서드를 설정하고 테스트하려면:

  1. /record를 선택한 다음 [메서드 생성] 을 선택합니다.

  2. 메서드 유형에서 PUT을 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 PutRecord를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

  13. 통합 요청 섹션에서 다음 URL 요청 헤더 파라미터를 추가합니다.

    Content-Type: 'x-amz-json-1.1'

    작업 절차는 GET /streams 메서드에 대한 요청 파라미터 매핑을 설정하는 절차와 같습니다.

  14. 다음의 본문 매핑 템플릿을 추가하여 데이터를 PUT /streams/{stream-name}/record 메서드 요청에서 POST /?Action=PutRecord의 해당 통합 요청으로 매핑합니다.

    {
    "StreamName": "$input.params('stream-name')",
    "Data": "$util.base64Encode($input.json('$.Data'))",
    "PartitionKey": "$input.path('$.PartitionKey')"
}

    이 매핑 템플릿에서는 메서드 요청 페이로드의 형식이 다음과 같다고 가정합니다.

    {
   "Data": "some data",
   "PartitionKey": "some key"
}

    이 데이터는 다음의 JSON 스키마에 의해 모델링할 수 있습니다.

    {
  "$schema": "http://json-schema.org/draft-04/schema#",
  "title": "PutRecord proxy single-record payload",
  "type": "object",
  "properties": {
      "Data": { "type": "string" },
      "PartitionKey": { "type": "string" }
  }
}

    이 스키마를 포함하는 모델을 생성한 다음 그 모델을 사용하면 매핑 템플릿을 용이하게 생성할 수 있습니다. 하지만 모델을 사용하지 않고 매핑 템플릿을 생성할 수도 있습니다.

  15. PUT /streams/{stream-name}/record 메서드를 테스트하려면 stream-name 경로 변수를 기존 스트림 이름으로 설정하고, 필수 형식의 페이로드를 공급한 다음, 메서드 요청을 제출합니다. 테스트가 성공하면 다음 형식의 페이로드가 포함된 200 OK 응답이 반환됩니다.

    
{
  "SequenceNumber": "49559409944537880850133345460169886593573102115167928386",
  "ShardId": "shardId-000000000004"
}
Kinesis에서 PUT /streams/{stream-name}/records를 호출할 수 있도록 PutRecords 메서드를 설정하고 테스트하려면

  1. /records 리소스를 선택한 다음 생성 방법을 선택합니다.

  2. 메서드 유형에서 PUT을 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 PutRecords를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

  13. 통합 요청 섹션에서 다음 URL 요청 헤더 파라미터를 추가합니다.

    Content-Type: 'x-amz-json-1.1'

    작업 절차는 GET /streams 메서드에 대한 요청 파라미터 매핑을 설정하는 절차와 같습니다.

  14. 다음의 매핑 템플릿을 추가하여 데이터를 PUT /streams/{stream-name}/records 메서드 요청에서 POST /?Action=PutRecords의 해당 통합 요청으로 매핑합니다.

    {
    "StreamName": "$input.params('stream-name')",
    "Records": [
       #foreach($elem in $input.path('$.records'))
          {
            "Data": "$util.base64Encode($elem.data)",
            "PartitionKey": "$elem.partition-key"
          }#if($foreach.hasNext),#end
        #end
    ]
}

    이 매핑 템플릿에서는 메서드 요청 페이로드를 다음의 JSON 스키마에 의해 모델링할 수 있다고 가정합니다.

    {
  "$schema": "http://json-schema.org/draft-04/schema#",
  "title": "PutRecords proxy payload data",
  "type": "object",
  "properties": {
    "records": {
      "type": "array",
      "items": {
        "type": "object",
        "properties": {
          "data": { "type": "string" },
          "partition-key": { "type": "string" }
        }
      }
    }
  }
}

    이 스키마를 포함하는 모델을 생성한 다음 그 모델을 사용하면 매핑 템플릿을 용이하게 생성할 수 있습니다. 하지만 모델을 사용하지 않고 매핑 템플릿을 생성할 수도 있습니다.

    이 자습서에서는 약간 다른 두 가지 페이로드 형식을 사용하여 API 개발자가 백엔드 데이터 형식을 클라이언트에 표시하거나 클라이언트로부터 숨길 수 있음을 보여 줬습니다. 형식 하나는 PUT /streams/{stream-name}/records 메서드(위)에 사용되고, 다른 형식은 PUT /streams/{stream-name}/record 메서드(이전 절차)에 사용됩니다. 프로덕션 환경에서는 두 형식의 일관성을 유지해야 합니다.

  15. PUT /streams/{stream-name}/records 메서드를 테스트하려면 stream-name 경로 변수를 기존 스트림으로 설정하고, 다음의 페이로드를 공급한 다음, 메서드 요청을 제출합니다.

    
{
    "records": [
        {
            "data": "some data",
            "partition-key": "some key"
        },
        {
            "data": "some other data",
            "partition-key": "some key"
        }
    ]
}

    테스트가 성공하면 다음과 출력과 비슷한 페이로드를 포함하는 200 OK 응답이 반환됩니다.

    {
  "FailedRecordCount": 0,
  "Records": [
    {
      "SequenceNumber": "49559409944537880850133345460167468741933742152373764162",
      "ShardId": "shardId-000000000004"
    },
    {
      "SequenceNumber": "49559409944537880850133345460168677667753356781548470338",
      "ShardId": "shardId-000000000004"
    }
  ]
}
Kinesis에서 GET /streams/{stream-name}/sharditerator를 호출할 수 있도록 GetShardIterator 메서드를 설정하고 테스트하려면

GET /streams/{stream-name}/sharditerator 메서드는 GET /streams/{stream-name}/records 메서드를 호출하기 전에 필수 샤드 반복자를 취득하는 헬퍼 반복자입니다.

  1. /sharditerator 리소스를 선택한 다음, 메서드 생성을 선택합니다.

  2. 메서드 유형에서 GET을 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 GetShardIterator를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

    GetShardIterator 작업은 ShardId 값의 입력이 요구됩니다. 클라이언트에서 제공한 ShardId 값을 전달하려면 다음 단계에 나온 것처럼 shard-id 쿼리 파라미터를 메서드 요청에 추가합니다.

  13. 메서드 요청 탭의 메서드 요청 설정에서 편집을 선택합니다.

  14. URL 쿼리 문자열 파라미터를 선택하고 다음을 수행합니다.

    1. 쿼리 문자열 추가(Add query string)를 선택합니다.

    2. 이름에 shard-id를 입력합니다.

    3. 필수 상태로 유지하고 캐싱을 해제합니다.

    4. 저장을 선택합니다.

    다음의 매핑 템플릿에서 shard-id 쿼리 파라미터 값을 Kinesis에서의 GetShardIterator 작업에 대한 입력으로 JSON 페이로드의 ShardId 속성 값으로 설정합니다.

  15. 통합 요청 섹션에서, 메서드 요청의 shard-id 및 stream-name 파라미터에서 GetShardIterator 작업에 대한 필수 입력(ShardId 및 StreamName)을 생성하도록 다음과 같은 매핑 템플릿을 추가합니다. 이에 더해, 매핑 템플릿은 기본값으로 ShardIteratorType을 TRIM_HORIZON로 설정합니다.

    {
    "ShardId": "$input.params('shard-id')",
    "ShardIteratorType": "TRIM_HORIZON",
    "StreamName": "$input.params('stream-name')"
}

  16. API Gateway 콘솔에서 테스트(Test) 옵션을 사용하여 기존 스트림 이름을 stream-name 경로(Path) 변수 값으로 입력하고, shard-id 쿼리 문자열(Query string)을 기존 ShardId 값(예: shard-000000000004)으로 설정한 다음, 테스트(Test)를 선택합니다.

    올바른 응답 페이로드는 다음 출력과 유사합니다.

    {
  "ShardIterator": "AAAAAAAAAAFYVN3VlFy..."
}

    [ShardIterator] 값을 기록해 둡니다. 스트림에서 레코드를 가져올 때 필요합니다.

Kinesis에서 GET /streams/{stream-name}/records 작업을 호출할 수 있도록 GetRecords 메서드를 구성하고 테스트하려면

  1. /records 리소스를 선택한 다음 생성 방법을 선택합니다.

  2. 메서드 유형에서 GET을 선택합니다.

  3. 통합 유형에서 AWS 서비스를 선택합니다.

  4. 에서 Kinesis 스트림을 생성한 AWS 리전 위치를 선택합니다. AWS 리전

  5. AWS 서비스에서 Kinesis를 선택합니다.

  6. AWS 하위 도메인은 비워 둡니다.

  7. HTTP 메서드에 대해 POST를 선택합니다.

  8. 작업 유형에서 사용자 작업 이름을 선택합니다.

  9. 함수 이름에 GetRecords를 입력합니다.

  10. 실행 역할에 실행 역할의 ARN을 입력합니다.

  11. 콘텐츠 처리에서 패스스루의 기본값을 유지합니다.

  12. 메서드 생성을 선택합니다.

    GetRecords 작업은 ShardIterator 값의 입력이 요구됩니다. 클라이언트에서 제공한 ShardIterator 값을 전달하려면 다음 단계에 나온 것처럼 Shard-Iterator 헤더 파라미터를 메서드 요청에 추가합니다.

  13. 메서드 요청 탭의 메서드 요청 설정에서 편집을 선택합니다.

  14. HTTP 요청 헤더를 선택하고 다음을 수행합니다.

    1. 헤더 추가(Add header)를 선택합니다.

    2. 이름에 Shard-Iterator를 입력합니다.

    3. 필수 상태로 유지하고 캐싱을 해제합니다.

    4. 저장을 선택합니다.

  15. 통합 요청 섹션에서 다음의 본문 매핑 템플릿을 설정하여 Shard-Iterator 헤더 파라미터 값을 Kinesis에서의 GetRecords 작업에 대한 JSON 페이로드의 ShardIterator 속성 값에 매핑합니다.

    {
    "ShardIterator": "$input.params('Shard-Iterator')"
}

  16. API Gateway 콘솔에서 테스트 옵션을 사용하여 기존 스트림 이름을 stream-name 경로 변수 값으로 입력하고, Shard-Iterator 헤더를 GET /streams/{stream-name}/sharditerator 메서드(위)의 테스트 실행에서 얻은 ShardIterator 값으로 설정한 다음, 테스트를 선택합니다.

    올바른 응답 페이로드는 다음 출력과 유사합니다.

    {
  "MillisBehindLatest": 0,
  "NextShardIterator": "AAAAAAAAAAF...",
  "Records": [ ... ]
}