[AWS Streaming] UI/UX Click Heatmap with AWS KDS, MSF, Glue, and Athena Pipeline

본 포스팅에선 실습을 위한 애플리케이션 및 (필요 시) Terraform 코드를 아래의 깃허브 레포지토리에서 제공한다. (단, 이 포스팅에선 Terraform을 사용하지 않고 AWS 콘솔을 통해 리소스를 생성해볼 것이다.)

https://github.com/yulmwu/aws-click-heatmap-demo

0. Overview

필자는 프론트엔드 개발자도 아니고 그쪽 분야로 관심이 있는 것 또한 아니다. 하지만 작은 팀이나 1인 개발 시 어쩔 수 없이 접해야 하고, 또한 최종적으로 실 사용자들에게 보여지는 분야 중 하나이기 때문에 최소한의 학습을 해야 하는 분야가 아닌가 싶다.

팀에서 최근 얼떨결에 프론트엔드와 UI/UX를 담당할 새로운 팀원과 함께하게 되었는데, 여러 이야기를 나눠보던 중 _"UX 개선을 위해 히트맵을 구현해보자."_라는 의견을 나누게 되었다.

---

---

그리고 이 포스팅에서는 Kafka 대신 AWS의 데이터 스트리밍 플랫폼인 KDS_(Kinesis Data Streams)_와 AWS에서 매니지드로 관리하는 Flink인 MSF_(AWS Managed Service for Apache Flink)_, S3 쿼리를 위한 Glue 및 Athena를 사용하여 PoC/MVP 정도의 페이지 클릭 히트맵을 구현해볼 것이다. 최종적으로 확인해볼 수 있는 히트맵은 아래와 같다.

이 포스팅에선 AWS 인프라를 다루며, 클릭 시 데이터가 Kinesis Data Streams에 쌓이도록 하는 애플리케이션(이하 Click Producer)과 Athena 쿼리 결과를 응답 받아 히트맵을 렌더링하는 애플리케이션(이하 Heatmap Viewer)에 대한 소스코드 설명은 하지 않는다.

다만 환경 변수 설정 및 빌드/실행 방법만 기술하고, 자세한 소스코드는 제공된 깃허브 레포지토리를 참고하자. PoC 수준이기 때문에 프로덕션 환경에서는 사용하지 않는 것을 극히 권장한다.

1. AWS Architecture

이 포스팅에서 구현해볼 AWS 아키텍처는 아래와 같다. 필자가 설계한 아키텍처라 Best Practice는 아닐 수 있지만, 주어진 상황에 가장 적합하다고 생각되었다.

---

아키텍처의 흐름은 아래와 같다. 각 서비스의 자세한 내용은 추후 다시 다루도록 하겠다.

1. Click Producer가 AWS Kinesis SDK PutRecordCommand를 통해 Kinesis Data Streams(KDS)에 레코드를 보낸다. (A)
2. Kinesis Data Streams는 2가지 서비스와 연결되는데, MSF 및 Kinesis Data Firehose로 파이프라인이 이어진다. (B)

(A) Kinesis Data Firehose

Kinesis Data Firehose는 Raw 데이터, 즉 Kinesis Data Streams로 부터 데이터를 컨슈밍하여 원본(Raw) 그대로를 S3에 저장한다. 이때 S3를 데이터 레이크로 사용하는 아키텍처인 것이다.

현재 아키텍처에선 Raw 데이터를 그대로 저장하는 것 외엔 Firehose를 Buffering, Lambda Transform 등의 다른 용도로 사용하지는 않는다.

추가적으로, 이 포스팅에선 MSF와 Firehose가 동일한 샤드의 읽기 용량을 나눠쓰는 Shared Throughput 방식을 사용하고, 규모가 커질 경우 EFO(Enhanced Fan-Out)를 고려해볼 수 있다.

추가적으로, 이 포스팅에선 KDS의 기본 커슈밍 모델인 Shared Throughput(기본 2MB/s 읽기) 방식을 사용한다. 즉 MSF와 Firehose가 동일한 샤드의 읽기 처리량을 공유하며 데이터를 컨슈밍하는데, 트래픽 규모가 커질 경우 Enhanced Fan-Out(EFO)를 고려해볼 수 있다.

(B) Amazon Managed Service for Apache Flink

MSF(Amazon Managed Service for Apache Flink)는 Raw 데이터를 큐레이션한다. (이때 Window, S3 FileSink, ParquetWriter 등의 기능이 사용된다. 이는 추후 따로 다루겠다.)

MSF(Flink)를 거친 큐레이션된 데이터는 Raw 데이터와 마찬가지로 S3(데이터 레이크) 버킷에 저장된다. 아키텍처에선 옵션으로 DynamoDB를 Flink 다음으로 붙였는데, 이는 실시간 처리를 위함이다.

이때의 Flink 애플리케이션이 특정 주기(TumblingProcessingTimeWindows, 1분으로 지정)마다 데이터를 Window로 나눠 추후 S3 Athena 쿼리 시 특정 시간을 범위로 하는 쿼리문을 작성해볼 수 있다. (본 예제에서는 Event Time 기반 Tumbling Window를 사용하며, Late Event 및 Watermark 처리는 단순화를 위해 고려하지 않았다.)

또한 원본에서는 프론트엔드의 뷰포트가 전부 다르기 찍히기 때문에, 이를 특정 크기의 Grid(20x20으로 지정)로 변환하여 히트맵 뷰어에서 처리한다. (단, 이 동작은 구현하기 나름이다. 필자는 이러한 방식으로 구현했지만, 실제 Flink 애플리케이션은 다를 수 있다.)

마지막으로 MSF(Flink) 애플리케이션은 S3 버킷에 데이터를 저장하고(FileSink), Athena 분석에서 최적화를 위해 Parquet 포맷으로 저장한다. (ParquetWriters)

Parquet 포맷은 주로 빅데이터, 하둡 생태계에서 많이 사용하는 컬럼 기반의 파일 포맷이다.

행(Row)이 아닌 열(컬럼) 단위로 저장하기 때문에 압축 효율이 높고, 쿼리를 통한 분석 시 데이터 처리 성능에 대해 최적화 할 수 있다.

대부분의 다양한 처리 엔진 및 분석 서비스에서 호환되는데, AWS Athena 및 Glue에서도 호환되는 포맷이기 때문에 사용하였다.

이렇게 큐레이션된 데이터가 Curated S3 버킷에 저장이 되었다면, AWS Glue Crawler를 통해 Athena를 위한 테이블 스키마와 파티션 정보를 Glue Data Catalog로 만든다. (Curated S3엔 dt='yyyy-MM-dd'/hour='HH" 형태로 저장됨)

Glue Crawler를 통해 자동으로 테이블의 스키마를 알 수 있고, 파티션(S3 디렉토리 경로)을 자동으로 찾을 수 있다. Glue는 30분 간격으로 스케쥴을 설정하여 스키마 및 파티션 정보를 Catalog로 생성할 수 있도록 한다.

마지막으로 Athena는 Glue를 통해 생성된 테이블의 스키마를 활용하여 Curated S3 버킷을 분석 한다. (쿼리) 쿼리 결과는 Athena Results S3 버킷에 따로 저장되며, 최종적으로 이 버킷에 저장된 쿼리 결과를 Heatmap Viewer 애플리케이션에서 화면에 렌더링한다. (단, 실시간 처리를 위함이라면 이 아키텍처 보다는 DynamoDB, Redis 등을 사용하여 서빙 스토어를 구축하는 것이 더 이득일 수 있다.)

---

설명을 길게 하였는데, 각 서비스의 자세한 동작 과정 등은 아키텍처를 구현하면서 개별적으로 설명하도록 하고 다음으로 아키텍처를 구현해보도록 하겠다.

2. Demo

실습에선 컴퓨팅 기능을 사용하지 않고, 매니지드 + 서비리스 조합으로 구성을 하였다. 이 중 Kinesis Data Streams (KDS)는 샤드 단위 시간으로 과금되고, Managed Service for Apache Flink (MSF)는 애플리케이션(Job)이 실행되는 시간 만큼 과금이 되므로 실습 시 요금 발생에 주의하길 바란다.

(1) S3 Bucket

버킷의 이름은 다음과 같다. 중복이 되면 안되니 겹친다면 변경하도록 하자.

• Raw S3 Bucket: heatmap-demo-raw-1230
• Curated S3 Bucket: heatmap-demo-curated-1230
• Athena Results S3 Bucket: heatmap-demo-athena-results-1230

---

그 외의 구성은 따로 하지 않아도 된다. S3 버킷에 접근할 땐 IAM 정책을 구성하여 접근할 것이다.

(2) Kinesis Data Streams

Kinesis Data Streams는 앞서 말했 듯 EFO를 구성하지는 않겠다. 다만 필요 시 구성하면 될 것이다. 또한 프로비저닝 모드 시 샤드 별로 활성화 시간으로 요금이 청구되니 참고하자.

---

필요에 따라 용량 모드로 가변적으로 동작하는 온디맨드를 선택할 수 있지만, 트래픽 양이 일정하다면 프로비저닝 모드가 더 이득일 수 있다. 필자는 프로비저닝 모드에 샤드의 수는 1개로 구성하도록 하겠다.

그 외의 옵션은 따로 설정하지 않겠다. Retention 기간 변경 등은 위와 같이 KDS가 활성화된 뒤 변경할 수 있다.

(3) Kinesis Data Firehose

IAM Role

현재 아키텍처에서 Kinesis Data Firehose는 S3 버킷 Write 권한과 Kinesis Data Streams에 대한 Read 권한이 필요하다.

먼저 역할을 만드는데, 엔티티 유형은 AWS 서비스 — Firehose로 선택하고, 권한 추가는 넘어가자. 역할 이름은 heatmap-demo-firehose-role로 설정하였다. 신뢰 정책은 아래와 같다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "firehose.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

그리고 역할에 붙일 아래와 같은 정책을 만들 것이다. 여기서 <...> 안의 내용은 본인이 구성한 서비스 이름에 맞게 수정해야 한다. 정책 이름은 heatmap-demo-firehose-policy이다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3Write",
      "Effect": "Allow",
      "Action": [
        "s3:AbortMultipartUpload",
        "s3:GetBucketLocation",
        "s3:GetObject",
        "s3:ListBucket",
        "s3:ListBucketMultipartUploads",
        "s3:PutObject"
      ],
      "Resource": [
        "arn:aws:s3:::heatmap-demo-raw-1230",
        "arn:aws:s3:::heatmap-demo-raw-1230/*"
      ]
    },
    {
      "Sid": "KinesisRead",
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamSummary",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords",
        "kinesis:ListShards"
      ],
      "Resource": "arn:aws:kinesis:ap-northeast-2:<ACCOUNT_ID>:stream/heatmap-demo-kds"
    }
  ]
}

그리고 아래와 같이 역할에 정책을 Attach하면 된다.

Kinesis Data Firehose

Firehose의 소스는 KDS(heatmap-demo-kds)이며, 대상은 S3를 선택한다.

옵션 중 Lambda를 사용하여 데이터를 가공하고 조작, 변경할 수 있는 기능 등이 있으나 생략한다.

대상 설정은 위와 같이 Raw S3 Bucket를 선택하고, S3 버킷 접두사는 raw/로 설정해두었다.

버퍼 크기는 기본값인 5MB로 설정해두었고, 필요에 따라 늘리면 된다. 또한 버퍼 간격은 300초로, 원본(Raw) 데이터는 굳이 빠르게 쌓이지 않아도 되므로 적절히 설정해두자. 압축은 사용하지 않았다.

IAM 역할은 위와 같이 만들어둔 heatmap-demo-firehose-role을 선택해주었다.

(4) Managed Service for Apache Flink (MSF)

Building Flink Artifacts (Jar)

MSF를 구성하기 전, MSF(Kafka)에 올릴 애플리케이션을 빌드하고 Jar 아키팩트를 S3(heatmap-demo-curated-1230) 버킷에 업로드하자. 애플리케이션은 아래의 깃허브 레포지토리의 applications/flink-heatmap-job 디렉토리에 위치한다.

git clone https://github.com/yulmwu/aws-click-heatmap-demo.git
cd aws-click-heatmap-demo/applications/flink-heatmap-job
 
mvn clean package
 
cd target
zip flink-heatmap-job-1.0.2.zip flink-heatmap-job-1.0.2.jar
 
# S3 경로는 s3://heatmap-demo-curated-1230/artifacts/heatmap-demo-flink로 구성하였으나, 변경해도 괜찮다.
aws s3 cp ./flink-heatmap-job-1.0.2.zip s3://heatmap-demo-curated-1230/artifacts/heatmap-demo-flink/flink-heatmap-job-1.0.2.zip

애플리케이션에서 필요한 환경 변수(프로퍼티)는 MSF 구성에서 설정할 수 있는데, KINESIS_STREAM_ARN, CURATED_S3_PATH, AWS_REGION 프로퍼티를 요구한다.

IAM Role

Firehose와 마찬가지로 IAM 역할과 역할에 Attach할 정책을 만들어줘야 한다. (heatmap-demo-msf-role, heatmap-demo-msf-policy) 각 신뢰 정책과 정책의 JSON은 아래와 같다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "kinesisanalytics.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

Kinesis Analytics는 Managed Service for Apache Flink(MSF)의 이전 이름이다. 즉 MSF이다.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3Access",
      "Effect": "Allow",
      "Action": [
        "s3:AbortMultipartUpload",
        "s3:GetObject",
        "s3:ListBucketMultipartUploads",
        "s3:PutObject",
        "s3:ListBucket",
        "s3:DeleteObject",
        "s3:ListMultipartUploadParts"
      ],
      "Resource": [
        "arn:aws:s3:::heatmap-demo-curated-1230",
        "arn:aws:s3:::heatmap-demo-curated-1230/*"
      ]
    },
    {
      "Sid": "KinesisRead",
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamSummary",
        "kinesis:GetRecords",
        "kinesis:GetShardIterator",
        "kinesis:ListShards",
        "kinesis:ListStreams",
        "kinesis:SubscribeToShard"
      ],
      "Resource": "arn:aws:kinesis:ap-northeast-2:<ACCOUNT_ID>:stream/heatmap-demo-kds"
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:PutLogEvents",
        "logs:CreateLogStream",
        "logs:DescribeLogStreams"
      ],
      "Resource": "arn:aws:logs:*:*:log-group:/aws/kinesis-analytics/heatmap-demo-flink:*"
    },
    {
      "Sid": "CloudWatchLogsCreateGroup",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup"
      ],
      "Resource": "*"
    }
  ]
}

아직 MSF 애플리케이션을 생성하지는 않았으나, 애플리케이션의 이름은 heatmap-demo-flink를 넣어두면 된다. 원할 경우 애플리케이션의 이름을 바꿀 수 있다.

Cloudwatch Log Group

이 부분은 옵션이긴 한데, MSF 애플리케이션 실행 시 발생하는 에러를 디버깅하기 위해 Cloudwatch 로그 그룹을 구성할 것이다. 로그 그룹의 이름은 /aws/kinesis-analytics/heatmap-demo-flink이다.

그리고 해당 로그 그룹 안에 application 로그 스트림을 생성해주자.

확인해보니 MSF 생성 시 자동으로 Cloudwatch 로그 그룹 및 스트림을 만드는 옵션이 있다. 이 방식을 사용해도 무방하다.

MSF

MSF 애플리케이션의 이름은 heatmap-demo-flink, Flink 버전은 1.18(FLINK-1_18)을 선택하였다. 포스팅을 작성하는 시점의 가장 최신 버전인 1.20을 선택해도 문제는 없겠지만, 1.18을 기준으로 Flink 애플리케이션을 작성하였기 때문에 1.18로 선택하였다.

IAM 역할도 Firehose와 마찬가지로 만들어뒀던 역할을 사용한다.

일단은 위와 같이 구성하고, Flink 애플리케이션을 만든다. 그리고 최소한의 요금 발생을 위한 Flink 설정과 애플리케이션 코드, 환경 변수 등을 설정해야한다.

애플리케이션 코드의 위치는 위와 같이 Jar 파일이 업로드된 S3 버킷의 위치과 경로를 입력한다. Jar 파일을 지정하라고 하는데, ZIP 파일을 사용해도 괜찮다.

그리고 병렬 처리(Parallelism) 옵션은 Parallelism = 1, Parallelism Per KPU = 1로 구성한다. 필요에 따라 늘리면 되지만, 최소한의 비용으로 실습해보기 위해 모두 1로 구성하였다.

그리고 Cloudwatch 옵션은 위와 같이 구성하였다. 에러나 경고만 확ㅇ니하려면 "경고"를 선택하면 된다.

마지막으로 런타임 속성은 위와 같이 구성하자. 아래의 값을 사용하면 된다.

Property Group = FlinkApplicationProperties (공통)

Key = KINESIS_STREAM_ARN
Value = arn:aws:kinesis:ap-northeast-2:<ACCOUNT_ID>:stream/heatmap-demo-kds

Key = CURATED_S3_PATH
Value = s3://heatmap-demo-curated-1230/curated/

Key = AWS_REGION
Value = ap-northeast-2

그 외의 옵션은 필요에 따라 구성하면 되고, 이와 같이 업데이트 후 Flink 애플리케이션을 실행하면 된다.

또는 아래와 같은 AWS CLI 명령어를 사용해도 된다. 실행이 완료되었다면 Cloudwatch에 로깅이 되어야 할 것이다.

aws kinesisanalyticsv2 start-application \
  --application-name heatmap-demo-flink \
  --run-configuration '{}'

(5) Athena Workgroup

Athena Workgroup은 아래와 같이 생성할 수 있다. 분석 엔진은 Athena SQL을 선택한다.

그리고 아래와 같이 Athena 쿼리 결과의 위치를 S3 버킷에 s3://heatmap-demo-athena-results-1230/athena-results/으로 구성한다.

(6) Glue Crawler

IAM

Glue Crawler를 위한 IAM 역할과 정책을 생성해주도록 하자. 이는 S3에 대한 권한과 Glue Catalog, CloudWatch에 대한 권한이 포함된다. (heatmap-demo-glue-role, heatmap-demo-glue-policy)

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "glue.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "S3ReadCurated",
      "Effect": "Allow",
      "Action": [
        "s3:GetObject",
        "s3:ListBucket"
      ],
      "Resource": [
        "arn:aws:s3:::heatmap-demo-curated-1230",
        "arn:aws:s3:::heatmap-demo-curated-1230/*"
      ]
    },
    {
      "Sid": "GlueCatalogAccess",
      "Effect": "Allow",
      "Action": [
        "glue:CreateTable",
        "glue:UpdateTable",
        "glue:GetDatabase",
        "glue:GetTable",
        "glue:GetTables",
        "glue:BatchGetPartition",
        "glue:BatchCreatePartition",
        "glue:CreatePartition",
        "glue:UpdatePartition",
        "glue:GetPartition",
        "glue:GetPartitions"
      ],
      "Resource": [
        "arn:aws:glue:*:*:catalog",
        "arn:aws:glue:*:*:database/heatmap_demo",
        "arn:aws:glue:*:*:table/heatmap_demo/*"
      ]
    },
    {
      "Sid": "CloudWatchLogs",
      "Effect": "Allow",
      "Action": [
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "arn:aws:logs:*:*:log-group:/aws-glue/crawlers:*"
    }
  ]
}

Glue 데이터베이스와 테이블의 이름은 heatmap_demo으로 미리 설정하였다. 마찬가지로 필요 시 변경하면 된다.

Glue Crawler

먼저 Glue 데이터베이스를 생성하자. 이 데이터베이스에 직접 데이터가 포함되는 것은 아니고, 스키마나 파티션 정보 등이 포함되는 것이다.

다음으로 Glue Crawler는 아래와 같이 구성한다.

데이터 소스는 S3로, 경로는 s3://heatmap-demo-curated-1230/curated/curated_heatmap/을 입력한다.

스케쥴링 설정은 cron(0/30 * * * ? *)으로, 30분 간격으로 Glue 크롤러가 실행될 수 있도록 한다.

이렇게 Glue Crawler를 구성하였다면, Glue Crawler를 실행하면 되지만 Curated S3에 데이터가 쌓여있는 상태에서 실행해야 정상적으로 동작하므로, 이는 추후 테스팅 단계에서 실행해보도록 하겠다.

(7) Applications

Click Producer와 Heatmap Viewer 애플리케이션 모두 마찬가지로 깃허브 레포지토리에서 확인해볼 수 있으며, 각각 applications 디렉토리의 heatmap-click-producer와 heatmap-athena-viewer 디렉토리에 위치한다.

Click Producer

Click Producer 애플리케이션에서 설정할 환경 변수(.env)는 아래와 같다. AWS Full Access Key를 직접 사용하도록 구성하였으니, 프로덕션 환경에선 권장하지 않는다.

VITE_AWS_REGION=ap-northeast-2
VITE_KINESIS_STREAM_NAME=heatmap-demo-kds
 
VITE_AWS_ACCESS_KEY_ID=...
VITE_AWS_SECRET_ACCESS_KEY=...
 
VITE_PAGE_ID=demo-page

아래의 명령어로 NPM 의존성을 설치하고 실행하자.

npm i
npm run start

Heatmap Viewer

Heatmap Viewer 애플리케이션에선 아래와 같은 환경 변수가 필요하다. 마찬가지로 Access Key를 제외한다면 (같은 리소스 이름으로 진행하였다는 가정 하에) 수정할 부분은 없다.

VITE_AWS_REGION=ap-northeast-2
 
VITE_ATHENA_WORKGROUP=heatmap-demo-wg
VITE_GLUE_DATABASE=heatmap_demo
VITE_GLUE_TABLE=curated_heatmap
 
VITE_AWS_ACCESS_KEY_ID=...
VITE_AWS_SECRET_ACCESS_KEY=...

npm i
npm run build
npm run start

3. Testing

만약 성공적으로 인프라 구축에 성공하고 두 애플리케이션의 환경 설정까지 완료하였다면 Click Producer 애플리케이션에 접속해보자.

여깃 Start sending 버튼을 클릭한 뒤 보이는 직사각형 영역 안을 여러 부분에 여러번 클릭해보자. 그럼 아래의 Response에 응답이 표시되어야 한다. Flink 애플리케이션에서 TumblingProcessingTimeWindows 시간을 1분으로 설정해두었기 때문에, 최소 1분이 지나야 결과를 확인해볼 수 있다. 조금만 기다린 다음, Curated S3 버킷을 확인해보자.

그럼 위와 같이 MSF(Flink) 애플리케이션에서 FileSink를 통해 저장된 Parquet 포맷의 오브젝트가 보일 것이다. (필자는 여러번 테스트를 하여 파일들이 많지만, 하나 밖에 없을 수도 있다.)

이 상태로 뷰어 애플리케이션을 실행하면 안되고, Glue Crawler를 실행하여 Athena가 사용할 스키마(테이블 스키마, 파티션 정보 등)를 만들어야 한다. AWS CLI를 통해서 실행할 수도 있고, 콘솔에서도 실행할 수 있다. (Glue Crawler를 특정 기간마다 실행하는 스케쥴러를 구성해두긴 하였으나, 그 시간만큼 기다리지 않고 직접 실행하는 것임을 참고하자.)

또는 아래의 CLI 명령어로 실행할 수 있다.

aws glue start-crawler --name heatmap-demo-curated-crawler
 
# 상태 확인
aws glue get-crawler --name heatmap-demo-curated-crawler
aws glue get-tables --database-name heatmap_demo # Glue 데이터베이스는 실제 데이터를 가지지 않고, 스키마나 파티션 정보와 같은 메타데이터 등을 카탈로그에서 관리한다. 실제 데이터는 S3나 RDS, DynamoDB 등에 저장된다.

Glue Crawler를 실행하고 나면 아래와 같이 스키마와 파티션 정보가 생성된 것을 볼 수 있다. 이제 Athena에서 이를 바탕으로 쿼리를 실행할 수 있다.

예시로 SELECT * FROM "heatmap_demo"."curated_heatmap"; 쿼리를 실행해보자. (Curated S3 내 모든 데이터를 가져오는 것이므로 사용 시 주의하도록 하자.)

이 결과는 아래와 같이 Athena Results S3 버킷에도 CSV 포맷으로 저장된다.

---

이렇게 Athena 쿼리를 실행해보고 Athena Results S3 버킷에 쿼리 결과가 저장되는 것을 확인하였다면, 마지막으로 히트맵을 렌더링하는 Heatmap Viewer 애플리케이션을 실행하고, 아래와 같은 Athena SQL 쿼리를 작성한 뒤 Run query 버튼을 클릭하자.

# AWS Athena는 Presto 엔진을 기반한다.
 
SELECT grid_x, grid_y, SUM(clicks) AS clicks
FROM "heatmap_demo"."curated_heatmap"
WHERE from_unixtime(window_end / 1000) >= date_add('hour', -1, current_timestamp)
GROUP BY 1, 2
ORDER BY clicks DESC;

그럼 위 사진과 같이 최근 1시간(date_add('hour', -1, current_timestamp)) 동안 집계된 클릭 데이터(Window)가 Grid 위에 히트맵으로 표시되는 것을 확인해볼 수 있다. (히트맵이 자연스럽지는 않는데, 그 부분의 코드는 생성형 AI가 작성하였다. 물론 이 실습에 있어 중요한 부분은 아니다.)

필자는 살짝의 시간을 간격으로 두고 여러번 클릭하였기 때문에 시간이 지나서 다시 조회해보면 위와 같이 최근 1시간 동안 집계된 Window가 표시되는 것을 확인해볼 수 있다. 만약 1시간이 아닌 30분으로 쿼리를 다시 보내본다면 아래와 같이 변화되는 것을 확인해볼 수 있을 것이다. (date_add('minute', -30, current_timestamp))

다음 날 다시 테스트(Click Producer + Heatmap Viewer)를 해보면 히트맵이 다르게 표시되는 것 또한 확인해볼 수 있을 것이다.

4. Monitoring, Observability

이 실습에서는 디버깅을 위해서 MSF(Flink)에 대해서만 Cloudwatch를 구성하였는데, 필요 시 Kinesis Data Streams나 Firehose에서도 적용해볼 수 있다. Cloudwatch를 사용하지 않고도 각 리소스 세부 정보에서 대략적인 메트릭을 확인해볼 수 있다.

(Kinesis Data Streams ::)

(Amazon Managed Service for Apache Flink ::)

또한 MSF는 매니지드 서비스답게 콘솔에서 대시보드 접속 버튼을 클릭하여 Flink 대시보드에 접속해볼 수 있다.

이상으로 포스팅을 마치겠다. 이 아키텍처를 기반으로 히트맵이 아니더라도 다양한 곳(특히 분석 쪽)에서 활용해볼 수 있으니 참고가 되었으면 좋을 것 같다.