Stream the data and validate output

Kinesis Data Stream에 데이터를 전송해서 S3 버킷으로 데이터가 저장되는지 검증해 보겠습니다.

1. Kinesis SDK나 KPL을 이용해서 Kinesis Data Stream으로 데이터를 전송을 시작하세요.

이전 실습에서 실행시킨 KPL 프로그램이 아직 구동중이면 다음 단계로 넘어가시면 됩니다.

2. Kinesis Producer가 Kinesis Data Stream으로 데이터 전송을 시작하면 해당 Kinesis Data Stream을 바라보고 있는 Kinesis Data Firehose Delivery Stream이 레코드를 수집 및 버퍼링하고, 람다 함수가 실행되서 데이터 변한 작업을 수행하고, 변환이 완료된 레코드는 목적지로 지정된 S3로 저장됩니다. Buffer는 이전 섹션에서 설정한 Buffer size 값인 128 MiB 또는 Buffer interval인 1분 중에 먼저 도달되는 순간 S3로 전달되서 저장됩니다. S3에 데이터가 표시되기까지 최대 1분이 소요될 수 있습니다.

3. Kinesis Data Stream으로 전송된 데이터가 S3에 저장될때까지 2-3분정도 기다려보겠습니다.

4. Kinesis 콘솔로 이동하여 nyc-taxi-trips Data Firehose Delivery Stream을 클릭합니다.

5. Monitoring 탭을 클릭하면 Firehose Delivery Stream의 일부 메트릭이 표시되는 것을 확인할수 있습니다.

6. Configuration 탭을 클릭합니다. Destination Settings 섹션이 보일때까지 아래로 스크롤합니다. S3 버킷 링크를 클릭합니다.

7. 몇분이 지나면 S3 버킷에 일부 데이터가 표시되어야 합니다. 그렇지 않다면 조금 더 기다려 주십시오. Parquet 형식의 데이터 파일이 보일 때까지 nyctaxitrips 폴더의 하위 폴더로 이동합니다.

8. Parquet 파일 중 하나를 선택하십시오. 그런 다음 Actions에서 Query with S3 Select를 선택합니다.

9. Input settings에서 Apache Parquet가 선택된 상태로 유지하고 Output settings에서 Format을 CSV로 선택한 다음 Run SQL Query을 클릭하세요.

10. 퀴리 실행이 완료되면 Query results탭 에서 CSV 형태로 불러온 데이터를 확인할수 있습니다.

11. Athena 콘솔로 이동합니다.

12. Editer 탭에서 Database 리스트에서 kinesislab을 선택하고 Tables 섹션 아래에 있는 nyctaxitrips 테이블 오른쪽에 있는 세로 줄임표 (⋮) 를 클릭해서 Load patitions를 클릭합니다.

13. nyctaxitrips 테이블 오른쪽에 있는 세로 줄임표 (⋮) 를 클릭하고 Preview table를 선택합니다.

14. 이렇게 하면 간단한 SQL 쿼리가 시작되며 완료되면 다음과 같습니다.

데이터가 표시되지 않으면 Partition Load 단계를 다시 실행하십시오.

15. 아래와 같은 쿼리를 실행해서 택시를 가장 많이 타는 시간대를 알아보겠습니다.

SELECT hour, count(hour) AS trips
FROM(
    SELECT 
    SUBSTRING(pickupdate, 12, 2) AS hour
    FROM "kinesislab"."nyctaxitrips"     
    )
GROUP BY hour
ORDER BY trips
DESC

16. Kinesis Data Stream으로 데이터를 전송하는 프로세스를 종료하세요.