Amazon DynamoDB용 Kinesis Data Streams 시작하기

AWS Management Console에 로그인하여 https://console.aws.amazon.com/kinesis/에서 Kinesis 콘솔을 엽니다.

Create data stream(데이터 스트림 생성)을 선택하고 지침에 따라 samplestream이라는 스트림을 생성합니다.

https://console.aws.amazon.com/dynamodb/에서 DynamoDB 콘솔을 엽니다.

콘솔 왼쪽의 탐색 창에서 테이블을 선택합니다.

Music 테이블을 선택합니다.

내보내기 및 스트림(Exports and streams) 탭을 선택합니다.

(선택 사항) Amazon Kinesis 데이터 스트림 세부 정보에서 레코드 타임스탬프 정밀도를 마이크로초(기본값)에서 밀리초로 변경할 수 있습니다.

드롭다운 목록에서 samplestream을 선택합니다.

켜기 버튼을 선택합니다.

create-stream command를 사용하여 samplestream이라는 Kinesis Data Streams를 생성합니다.

aws kinesis create-stream --stream-name samplestream --shard-count 3 

Kinesis 데이터 스트림의 샤드 수를 설정하기 전에 Kinesis Data Streams에 대한 샤드 관리 고려 사항 섹션을 참조하세요.

describe-stream command를 사용하여 Kinesis 스트림이 활성 상태이고 사용할 준비가 되어 있는지 확인합니다.

aws kinesis describe-stream --stream-name samplestream

DynamoDB enable-kinesis-streaming-destination 명령을 사용하여 DynamoDB 테이블에서 Kinesis 스트리밍을 활성화합니다. stream-arn 값은 이전 단계에 describe-stream에서 반환한 값으로 바꿉니다. 선택적으로 각 레코드에서 반환되는 타임스탬프 값을 마이크로초 단위로 더 정밀하게 설정하여 스트리밍을 활성화할 수 있습니다.

마이크로초의 타임스탬프 정밀도로 스트리밍 활성화:

aws dynamodb enable-kinesis-streaming-destination \
  --table-name Music \
  --stream-arn arn:aws:kinesis:us-west-2:12345678901:stream/samplestream
  --enable-kinesis-streaming-configuration ApproximateCreationDateTimePrecision=MICROSECOND

또는 기본 타임스탬프 정밀도(밀리초)로 스트리밍 활성화:

aws dynamodb enable-kinesis-streaming-destination \
  --table-name Music \
  --stream-arn arn:aws:kinesis:us-west-2:12345678901:stream/samplestream

DynamoDB describe-kinesis-streaming-destination 명령을 사용하여 테이블에서 Kinesis 스트리밍이 활성 상태인지 확인합니다.

aws dynamodb describe-kinesis-streaming-destination --table-name Music

DynamoDB 개발자 안내서에 설명된 대로 put-item 명령을 사용하여 DynamoDB 테이블에 데이터를 씁니다.

aws dynamodb put-item \
    --table-name Music  \
    --item \
        '{"Artist": {"S": "No One You Know"}, "SongTitle": {"S": "Call Me Today"}, "AlbumTitle": {"S": "Somewhat Famous"}, "Awards": {"N": "1"}}'

aws dynamodb put-item \
    --table-name Music \
    --item \
        '{"Artist": {"S": "Acme Band"}, "SongTitle": {"S": "Happy Day"}, "AlbumTitle": {"S": "Songs About Life"}, "Awards": {"N": "10"} }'

Kinesis get-records CLI 명령을 사용하여 Kinesis 스트림 콘텐츠를 검색합니다. 그 후 다음 코드 조각을 사용하여 스트림 콘텐츠를 역직렬화합니다.

/**
 * Takes as input a Record fetched from Kinesis and does arbitrary processing as an example.
 */
public void processRecord(Record kinesisRecord) throws IOException {
    ByteBuffer kdsRecordByteBuffer = kinesisRecord.getData();
    JsonNode rootNode = OBJECT_MAPPER.readTree(kdsRecordByteBuffer.array());
    JsonNode dynamoDBRecord = rootNode.get("dynamodb");
    JsonNode oldItemImage = dynamoDBRecord.get("OldImage");
    JsonNode newItemImage = dynamoDBRecord.get("NewImage");
    Instant recordTimestamp = fetchTimestamp(dynamoDBRecord);

    /**
     * Say for example our record contains a String attribute named "stringName" and we want to fetch the value
     * of this attribute from the new item image. The following code fetches this value.
     */
    JsonNode attributeNode = newItemImage.get("stringName");
    JsonNode attributeValueNode = attributeNode.get("S"); // Using DynamoDB "S" type attribute
    String attributeValue = attributeValueNode.textValue();
    System.out.println(attributeValue);
}

private Instant fetchTimestamp(JsonNode dynamoDBRecord) {
    JsonNode timestampJson = dynamoDBRecord.get("ApproximateCreationDateTime");
    JsonNode timestampPrecisionJson = dynamoDBRecord.get("ApproximateCreationDateTimePrecision");
    if (timestampPrecisionJson != null && timestampPrecisionJson.equals("MICROSECOND")) {
        return Instant.EPOCH.plus(timestampJson.longValue(), ChronoUnit.MICROS);
    }
    return Instant.ofEpochMilli(timestampJson.longValue());
}

Kinesis Data Streams 개발자 안내서의 지침에 따라 Java를 사용하여 samplestream이라는 Kinesis 데이터 스트림을 생성합니다.

Kinesis 데이터 스트림의 샤드 수를 설정하기 전에 Kinesis Data Streams에 대한 샤드 관리 고려 사항 단원을 참조하세요.

다음 코드 조각을 사용하여 DynamoDB 테이블에서 Kinesis 스트리밍을 활성화합니다. 선택적으로 각 레코드에서 반환되는 타임스탬프 값을 마이크로초 단위로 더 정밀하게 설정하여 스트리밍을 활성화할 수 있습니다.

마이크로초의 타임스탬프 정밀도로 스트리밍 활성화:

EnableKinesisStreamingConfiguration enableKdsConfig = EnableKinesisStreamingConfiguration.builder()
  .approximateCreationDateTimePrecision(ApproximateCreationDateTimePrecision.MICROSECOND)
  .build();

EnableKinesisStreamingDestinationRequest enableKdsRequest = EnableKinesisStreamingDestinationRequest.builder()
  .tableName(tableName)
  .streamArn(kdsArn)
  .enableKinesisStreamingConfiguration(enableKdsConfig)
  .build();

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

또는 기본 타임스탬프 정밀도(밀리초)로 스트리밍 활성화:

EnableKinesisStreamingDestinationRequest enableKdsRequest = EnableKinesisStreamingDestinationRequest.builder()
  .tableName(tableName)
  .streamArn(kdsArn)
  .build();

EnableKinesisStreamingDestinationResponse enableKdsResponse = ddbClient.enableKinesisStreamingDestination(enableKdsRequest);

Kinesis Data Streams 개발자 안내서의 지침에 따라 생성한 데이터 스트림에서 읽어옵니다.

다음 코드 조각을 사용하여 스트림 콘텐츠를 역직렬화합니다.

/**
 * Takes as input a Record fetched from Kinesis and does arbitrary processing as an example.
 */
public void processRecord(Record kinesisRecord) throws IOException {
    ByteBuffer kdsRecordByteBuffer = kinesisRecord.getData();
    JsonNode rootNode = OBJECT_MAPPER.readTree(kdsRecordByteBuffer.array());
    JsonNode dynamoDBRecord = rootNode.get("dynamodb");
    JsonNode oldItemImage = dynamoDBRecord.get("OldImage");
    JsonNode newItemImage = dynamoDBRecord.get("NewImage");
    Instant recordTimestamp = fetchTimestamp(dynamoDBRecord);

    /**
     * Say for example our record contains a String attribute named "stringName" and we wanted to fetch the value
     * of this attribute from the new item image, the below code would fetch this.
     */
    JsonNode attributeNode = newItemImage.get("stringName");
    JsonNode attributeValueNode = attributeNode.get("S"); // Using DynamoDB "S" type attribute
    String attributeValue = attributeValueNode.textValue();
    System.out.println(attributeValue);
}

private Instant fetchTimestamp(JsonNode dynamoDBRecord) {
    JsonNode timestampJson = dynamoDBRecord.get("ApproximateCreationDateTime");
    JsonNode timestampPrecisionJson = dynamoDBRecord.get("ApproximateCreationDateTimePrecision");
    if (timestampPrecisionJson != null && timestampPrecisionJson.equals("MICROSECOND")) {
        return Instant.EPOCH.plus(timestampJson.longValue(), ChronoUnit.MICROS);
    }
    return Instant.ofEpochMilli(timestampJson.longValue());
}