Kafka 파이프라인에 따른 DB 부하줄이기 여정

이전글 읽기 > 

https://hajinnote.tistory.com/105

ksqlDB를 이용해 실시간 이벤트 스트리밍 최적화하기

배경

작년 B2G 사업 시연에 사용될 학생의 학습기록 대시보드를 실시간 반영이 되는 형태로 작업해달라는 요청이 왔다. 당시 마감일이 너무 급박해 최적화를 완전히 진행하지 못하고 빠르게 설계해서 개발하고 배포를 했다. 시연이 무사히 끝나고 본격적으로 사용하기 위해 운영 환경에도 배포를(2023/11) 했는데, 운영 쪽은 이용자 수가 훨씬 많다보니 이벤트 수가 폭발을 하였고 최적화가 미비했던 부분에서 문제가 생겼다. 컴퓨팅이 낮은 성능이 아님에도 피크 시간대에 CPU 사용량이 99%를 찍고 새벽에 해소되는 형태를 띄었다. 원인이 이것저것 많았기 때문에 하나씩 분석하고 찾아내서 점차 DB 부하를 줄여나가는 여정을 담은 글이 될 예정이다.  

파이프라인 구조

자세히는 못적지만 문제가 있던 파이프라인의 구조는 대략 다음과 같다. 

학습활동 발생시 id를 이벤트 발송 > [앞단 토픽] > 컨슈머에서 id 받아 관련 값 조회 후 저장 > 저장된 id의 단위별 집계를 위해 이벤트 발송 > [뒷단 토픽] > 각 컨슈머에서 단위별 집계 처리하여 저장 

여기서 학습활동에서 하나라도 업데이트가 일어나면 전부 이벤트를 발송하다보니 뒤의 단위별 컨슈머에서 병목 현상이 일어났다. 

첫 접근 - 상위쿼리 확인 

데이터 팀은 AWS의 RDS aurora postgreSQL을 주 운영 데이터베이스로 사용하고 있다. RDS에서 성능 모니터링과 상위 쿼리를 확인 할 수 있어서 부하를 주는 상위쿼리가 무엇인지부터 확인했다. 

1, 2, 3위 쿼리 전부 집계에 관련된 SELECT문 이었다. 각 컨텐츠나 챕터 같은 커다란 단위별로 해당 유저가 수행한 학습결과에 따라 집계를 해서 저장하는 부분이 있다. 문제는 여기서 사용하는 각 단위가 워낙 크고, 또한 선생님에 따라 몇만개의 문제를 넣어서 활용하는 선생님도 있다보니 몇몇개의 SELECT문이 몇십초씩 걸리는 것이었다. 실시간으로 빠르게 처리해야 하는데 하나의 쿼리에 몇십초가 걸리니 당연히 처리가 제대로 되지 않았다.

상위쿼리 분석에 따른 대응 방안  

1. 상위쿼리의 update&delete 문
 > 조건문와 JOIN문에 사용하는 칼럼인데 index가 누락되었던 곳을 찾아 index 추가하여 해결 

2. 쿼리 최적화
 > 서브쿼리를 빼고 파이썬에서 계산하게 하거나, 쿼리 자체의 최적화를 통해 DB 자체의 연산량 최소화 

쿼리 예시를 보자. 아래는 특별한 요구사항인 2회 이상 학습을 했는지 아닌지 여부를 알기 위해 자기 자신을 id2로 그룹바이한 다음 LEFT JOIN을 사용한 쿼리이다. 자세하게 적을 순 없지만 저 상위 id가 클수록 처리해야 할 값이 늘어나면서 쿼리가 몇 초씩 걸리는 경우도 생겼다.

SELECT
    id, 
    CASE WHEN fq.frequency > 1 THEN 1 ELSE 0 END AS is_over_two_frequency
  FROM activity 
  LEFT JOIN (select id, count(*) as frequency 
              from activity 
              where id2의 상위id = {}
              group by id2) fq
  on fq.id2 = activity.id2

변경 쿼리는 아래와 같다. 비효율적인 LEFT JOIN을 없애고 Partition by를 사용해 select 문에서 처리 할 수 있게 변경해 몇 초씩 걸리던 쿼리가 1초 정도로 줄어들었다. 

CASE
  WHEN COUNT(*) OVER (PARTITION BY id2) > 1
    THEN 1
    ELSE 0
END AS is_over_two_frequency,

3. 너무 많은 중복 이벤트 처리 
> 이 부분이 가장 큰 부하를 주는 부분이었는데 앞 부분에서는 하나라도 업데이트가 일어날 때마다 이벤트를 보낸다. 이를 뒤의 단위별 집계에서 모든 이벤트를 받아 처리하다보니 사실상 중복이벤트를 지속적으로 처리하는 꼴이 되었고 이 때문에 심한 부하가 일어났다. 큰 단위의 집계에서는 값이 한두개 바뀐다고 큰 티가 나지 않기때문에 실시간으로 모든 집계를 업데이트 해주지 않아도 된다. 

중복 이벤트 쓰로틀링 

단시간내의 너무 많은 중복 이벤트가 부하의 원인임을 파악하고 나서 컨슈머 앞단에 쓰로틀링을 구현하여 이벤트를 줄이기로 작업방향을 잡았다. 
** [쓰로틀링(Throttling): 시스템의 자원 사용량, 서버로의 요청 수, 데이터 전송 속도 등을 인위적으로 제한하는 기술이나 접근 방식]
여기서는 동일 이벤트가 너무 많은 집계단위로 흘러들어가는 것을 막기 위함으로, 시간이나 개수로 모아서 그 안에 있는 중복값을 날리고 unique한 이벤트만 남겨 컨슈머로 보내기로 했다. 

시간 테스트 

먼저 시간으로 모아서 처리하게 될 때 어떤 효과가 있는지 테스트를 진행했다. 이벤트가 가장 많이 발생하는 시간대인 저녁시간에 초 단위로 그룹화하여 중복을 제거한 후, 이벤트가 얼마나 감소하는지 확인해보았다. 10초까지는 거의 비슷한 감소량을 보이는데 10초 단위로 생성되서 들어오는 더 앞단의 이벤트가 있어서 10초 이하일때는 중복 이벤트 개수가 거의 비슷하기 때문이다. 30초에서 1/3로 중복이 감소하는 것을 확인하고 30초를 기준으로 잡았다. 

초 단위	중복 처리 전 이벤트 개수	중복 처리 후 이벤트 개수	감소량
1	1334	761	43% 감소
5	1978	931	53% 감소
10	1373	728	47% 감소
30	1749	669	61% 감소
60	7520	1149	84% 감소

ksqlDB 사용

처음에는 혹시나 복잡한 로직이 들어갈까봐 컨슈머 앞에 쓰로틀링을 처리하는 토픽을 하나 더 두려고 했는데, 카프카 스트림즈나 ksql을 사용해서 더 간단하게 처리 할 수 도 있다는 것을 보고 본격적으로 사용방법을 공부했다.

** [ksqlDB란?]
- kafka의 스트리밍 데이터를 처리하기 위한 스트림 처리 언어
- SQL과 유사한 문법을 사용해 실시간 데이터 스트림 처리를 간단하게 만듦
- kafka 토픽에서 스트림 데이터를 읽어 쿼리, 필터, 집계 등 작업 수행

ksql에서 window tumbling이라는 집계 쿼리를 사용하면 해당 윈도우 안에 있는 값들에 대한 처리를 할 수 있다. 아래 예시 쿼리를 보자. SIZE 로 윈도우 크기를 정할 수 있으며, id로 group by를 걸면 윈도우 내의 동일한 id는 전부 사라지고 unique한 이벤트 id만 남게 된다. 이렇게 unique ID를 컨슈머에서 소비하게 해주면 작업은 끝이다. 

CREATE TABLE UNIQUE_TBL AS
SELECT 
 id,
 COUNT(*) AS "count"
FROM stream
WINDOW TUMBLING (SIZE 30 SECONDS)
GROUP BY id

 

간단해보이지만 처음 사용해보는거여서 window 집계 쿼리에 대해 공식문서의 예시를 보며 공부하면서 진행했다. 그 중 특히 삽질했던 부분을 소개하자면, ksql은 group by 절에 넣은 칼럼은 저절로 토픽의 메세지 키가 되고, 메세지 키는 곧 파티션의 키가 된다. 즉 위의 쿼리 예시에서는 id가 파티션 키이기 때문에 동일 id는 동일 파티션으로 계속 들어가게 되는 것. (파티션 별 순서보장을 위해서 파티션 키가 중요함) 하지만 group by 절에 여러 칼럼을 넣으니까 멀티 칼럼은 키로 지정할 수 없다며 에러가 났다. 이 부분을 열심히 구글링하여 찾아냈는데 키 포맷을 정하는 옵션이 있어서 json이나 avro 같은 직렬화 포맷을 사용하면 멀티 칼럼을 통채로 직렬화하여 키로 지정하여 사용할 수 있었다. 예시는 아래와 같다. 

CREATE TABLE UNIQUE_TBL WITH (FORMAT='JSON') AS
SELECT 
 id,
 id2, 
 id3,
 COUNT(*) AS "count"
FROM stream
WINDOW TUMBLING (SIZE 30 SECONDS)
GROUP BY id, id2, id3

윈도우 사이즈는 미리 테스트 했던 시간대를 참고해서 작업했고, 최종 1분으로 결정해서 적용했다. 

위처럼 테이블을 생성하면 테이블의 값을 저장하는 토픽이 생성되는데 이제 컨슈머에서 해당 토픽을 소비 할 수 있게 코드에 가서 바라보는 토픽을 변경해주면 된다. 이렇게 ksql로 unique 이벤트만 추려서 보내는 작업을 앞단과 뒷단 토픽에 전부 적용했다. 컨슈머에서는 이제 중복이 제거된 unique한 이벤트만 읽고 처리하게 되므로 그동안의 너무 많은 이벤트를 처리하느라 걸렸던 과부하가 해소 될 것이다. 

효과

카프카 모니터링 도구로 datadog을 사용하고 있어서 이를 통해 이벤트 개수가 얼마나 줄었는지 확인해보자. 

적용 전 이벤트 개수

빨간 부분이 앞단의 토픽이고 초록 부분이 뒷단의 집계를 위한 토픽이다. 앞단의 토픽에서 값을 받아 windowing으로 중복을 처리하고 뒤의 토픽으로 보낸다. 그래서 두 토픽의 이벤트 개수 차이를 비교해보면 중복 처리가 얼마나 되었는지 알 수 있다. Sum 부분이 해당 기간 안에 보내진 이벤트의 총 개수이다. 적용 전에는 4.87M 개 이벤트를 받아서 4.84M 개를 보낸 것으로 확인이 되는데, 두 토픽간의 개수 차이가 거의 나지 않음을 알 수 있다. 

적용 후 이벤트 개수 

ksqlDB 적용 후 6.75M 개 이벤트를 받아서 2.18M 개를 보낸 것을 확인 할 수 있다. 이벤트 개수가 약 67% 정도 감소했으니 꽤나 성공적으로 중복이 처리되었다. 

적용 전 CPU

 

DB는 AWS RDS를 사용하고 있는데 모니터링으로 가서 CPU의 변화를 확인해볼 수 있다. 적용 전에는 99%로 피크를 치면서 새벽까지도 해소가 되지 않다가 새벽 6시에 줄어드는 것을 볼 수 있는데, 이미 뒷단 토픽이 밀려서 처리되고 있기 때문에 60%를 계속 차지하고 있다. 

적용 후 CPU

적용 후 피크 시간대에도 90%를 넘지 않게 되었고 새벽 2시 정도에 보다 빠르게 해소가 되는 것으로 확인했다. 무엇보다 밀려서 처리되느라 60%를 항상 잡고 있었던 부분이 해소가 되었다. 이 부분을 더 개선하려면 현재 뒷단 토픽인 단위별 집계 토픽에서 1분으로 해놓은 윈도우 사이즈를 더 늘리는 방안이 있다. 사실 단위별 집계는 일분 단위도 엄청 짧은 단위여서 30분이나 그 이상 늘려도 될 것 같다. 하지만 이 부분은 다음 프로젝트로 넘어가느라 아직 진행하진 못했고, 새로운 토픽이나 무언가 추가 될때 추가로 개선 작업을 진행하게 될 것 같다. 

회고 

이번 부하 줄이기 여정은 이 정도로 일단락 되었다.(여기서 기능 추가 등으로 더 문제가 생긴다면 다음 작업을 이어서 진행할 것 같다.) 어디서 제일 큰 효과가 있는지 체크하려고 단계별로 진행하느라 약 한달정도 소요된 것 같다. 마지막에 막혔던 이벤트가 사르르 사라질때 정말 속이 뻥뚫리는 기분이 들었다. production 환경에서는 이벤트가 많이 발생해서 실시간으로 처리에 문제가 생기지 않을까 했지만 당시에는 감이 잘 안왔었다. (여담으로 사실은 데이터팀의 입장에서는 집계 부분은 배치로 돌리고 싶었는데, 그렇다고 실시간성을 버릴수는 없다고 해서 사용하게 되긴했다..) 역시나 문제가 터지는 것을 보며 요구사항이 있어도 기술 선택시 트레이드 오프를 명확히 정리하고 전달하며 조율을 해야겠다고 느꼈다. 왜냐면 결국 시간단위로 이벤트를 모아서 처리하기 때문에 완전한 실시간은 아니게 된거니까.. 개발도 개발이지만 이런 운영적인 이슈와 유지보수가 정말 중요한 작업임을 다시 한번 깨닫는다. 아무튼 이번 작업에서는 한단계 한단계씩 부하가 줄어가는 걸 확인하는게 재밌었고 엄청 많은 부분을 배운 것 같다.