KINESIS 명령어 및 python API 정리

1. Kinesis data stream 생성

$ aws kinesis create-stream --stream-name {STREAM_NAME} --shard-count 1

2. data stream 정보 확인

$ aws kinesis describe-stream --stream-name {STREAM_NAME}

3. 스트림 리스트 출력

$ aws kinesis list-streams

4. 레코드 넣기

$ aws kinesis put-record --stream-name {STREAM_NAME} --partition-key {KEY_VALUE} --data testdata1

$ aws kinesis put-record --stream-name {STREAM_NAME} --partition-key {KEY_VALUE} --data testdata2

$ aws kinesis put-record --stream-name {STREAM_NAME} --partition-key {KEY_VALUE} --data testdata3

$ aws kinesis put-record --stream-name {STREAM_NAME} --partition-key {KEY_VALUE} --data testdata4

5. 레코드 가져오기 (shard iterator 가져오기)

$ aws kinesis get-shard-iterator --shard-id shardId-000000000000 --shard-iterator-type TRIM_HORIZON --stream-name {STREAM_NAME}

$ aws kinesis get-records --shard-iterator {ITERATOR_VALUE}

# ShardIteratorType

AT_SEQUENCE_NUMBER: 값(StartingSequenceNumber)에 제공된 특정 시퀀스 번호로 표시된 위치에서 읽기 시작합니다 .

AFTER_SEQUENCE_NUMBER: 특정 시퀀스 번호(StartingSequenceNumber)로 표시된 위치 바로 다음에 값을 제공하여 읽기 시작합니다.

AT_TIMESTAMP: 특정 시간(Timestamp)에 명시된 위치에서 값을 제공하여 읽기 시작합니다.

TRIM_HORIZON: 샤드에서 가장 오래된 데이터 레코드 인 시스템 샤드의 마지막 트리밍되지 않은 레코드에서 읽기 시작합니다.

LATEST: 샤드에서 가장 최근의 레코드 바로 다음에 읽기 시작하므로 항상 샤드에서 가장 최신 데이터를 읽을 수 있습니다.

6. 데이터 확인

$ echo "QWJjZDE="| base64 --decode

7. stream 삭제

$ aws kinesis delete-stream --stream-name {STREAM_NAME} 

8. 모니터링 enable

$ aws kinesis enable-enhanced-monitoring --shard-level-metrics ALL --stream-name {STREAM_NAME} 

# KINESIS SDK API 참고

https://www.arundhaj.com/blog/getting-started-kinesis-python.html

https://docs.aws.amazon.com/ko_kr/code-samples/latest/catalog/python-kinesis-kinesis_to_firehose_to_s3.py.html

https://blog.sqreen.com/streaming-data-amazon-kinesis/

https://aws.amazon.com/ko/blogs/big-data/snakes-in-the-stream-feeding-and-eating-amazon-kinesis-streams-with-python/

https://pypi.org/project/kinesis-producer/

https://blog.ruanbekker.com/blog/2019/06/22/play-with-kinesis-data-streams-for-free/