Druid를 활용한 광고 통계 개발 경험 공유
이번 글에서는 네이버의 성과형 Display-Ad Platform(GFA) 에서 경험한 광고 성과 지표 API를 개발한 방법에 대해서 공유를 드려보고자 합니다.
시작하면서..
저희 팀은 GFA 에서 로그 데이터의 처리를 수집하고, 저장 분석하여 통계 데이터를 제공하는 업무를 담당하고 있습니다.
기존의 Mysql 기반으로 통계 데이터를 제공하던 구조를 Druid 를 적용하여 개선한 내용에 대해 공유 드리려 합니다.
Druid 적용 이전 구성
위의 구성과 같이 GFA 의 로그는 서버를 통해 Kafka 로 저장이 되고, Kafka 로그를 기반으로 광고 노출,클릭,매출 등의 통계 데이터를 처리하고 있습니다.
사용자 제공 통계
수천 TPS 이상으로 유입되는 로그를 처리한 통계 데이터를 매번 DB에 저장할 수 없어서 kafka 로그를 Streaming 을 통해 Redis 에 저장하였습니다.
그리고 Redis 에 저장된 통계 데이터를 10분 주기 배치로 통해 Mysql DB 로 저장하였습니다.
Unique 이용자 지표를 제공하여야 해서 해당 정보를 streaming 으로 HBase 저장하였고, 저장된 결과에서 배치를 통해 Unique 이용자 수치를 추출하여 Mysql DB 로 저장하였습니다.
Mysql 에 시계열 데이터 및 누적 통계 데이터를 저장하여, 통계 데이터 제공에 사용하였습니다.
내부 운영자 제공 통계
내부 운영 통계는 kafka 로그를 es 로 저장하여 kibana 대시보드를 이용하였습니다.
그리고 kafka 로그를 flume 을 통해 hdfs 로 저장하였고, hive 쿼리를 배치로 수행하여 Mysql 로 저장하였고, 스팟 데이터 추출은 hive, spark sql 을 이용하였습니다.
통계 개선 요구사항
플랫폼이 고도화 되면서 아래와 같은 요청 사항을 구현을 하여야 했습니다.
- 기존에 10분 배치로 통계 데이터가 갱신이 되었는데, Realtime 비딩을 위해 실시간 으로 통계 제공
- 기존의 통계 테이블의 dimension 4개 였는데, 다차원 분석을 위해 30개 이상의 dimension 을 추가로 저장
- 다차원 지표에서 유니크 이용자 수 제공
- 그래프 위한 매출 상위 N개 광고 (Approximate Top-N)
- 지표 다운로드 ( excel, csv )
Druid 적용 결정
요구 사항을 충족하기 위해 검토해본 데이터 저장소 중 druid 가 기존 구조에 바로 적용하기 용이하고, 익숙한 환경을( hadoop-eco, kafka, sql ) 지원하고 있었습니다.
성능 검증을 위해 파일럿을 진행하였고, 만족할 성능이 확인되어 시계열 데이터의 저장소로 druid 를 사용하기로 하였습니다.
다만, druid 에서 시계열 데이터 저장만 가능하여, 누적 데이터는 Mysql 을 이용하고 있습니다.
적용 후 구조
적용 후 변경된 구조를 보시면, 기존과는 kafka 의 원본 로그에 통계 정보나 과금 정보를 추가하여 별도의 통계 로그를 생성하였습니다.
이렇게 별도의 토픽을 생성한 이유는 druid 에서 제공하는 streaming ingestion 중 kafka indexing service 을 사용하기 위해서 입니다.
kafka indexing service
druid 에서는 batch ingestion 과 Streaming ingestion 을 지원하는데, 실시간 지표 제공을 위해서는 Streaming ingestion 을 이용하여야 했고, kafka 의 data 를 exactly-once 를 지원하는 kafka index service 가 있어서 사용하였습니다.
druid 에 저장된 데이터를 일부만 갱신할 수 있는 방법이 없어서, 부분 수정의 경우 가능하다면 보정을 위한 kafka 로그를 이용하여 보정을 진행하였고,
대규모 수정이나, dimension , matric 추가 등의 이슈는 hadoop 에 저장된 원본 로그 파일을 이용하여 hadoop batch ingestion 을 이용하였습니다.
Kafka 로그에 필드 추가
streaming ingestion 을 위해 통계 로그를 위해 원본 로그에 필요한 dimension 과 matric 을 추가하였습니다.
- time dimension : druid 에는 시계열 데이터만 저장할 수 있어서 time 값이 필수
- 분석 dimension 중 log 에 없는 dimension : streaming 처리를 통해 로그를 추가
- metric 필드 : 대부분의 metric 은 sum 이라 sum 을 위한 수치 추가
로그 예시
// 원본 로그
{"creativeNo":123, "userNo":"user123", "actionType":"CLICK"}// 통계 로그 - time dim, metric 추가
{"action_time":"2019-10-10T00:00:00+0900","creativeNo":123, "userNo":"user123",
"actionType":"CLICK", "clickCount":1, "spend":22000, "unpaidSpend":10000}
데이터 수정
일부 데이터 보정 — 로그 추가
Druid 는 부분 수정이 용이하지 않은 구조이고, streaming ingestion 을 사용 하기에 더욱 수정이 용이하지 않았습니다.
저희의 경우, 대부분의 보정이 수치를 수정하는 것이고, sum 기반의 데이터이기에 집계 데이터 수정을 위해 마이너스 로그를 kafka 추가하여 대응하였습니다.
“API 버그로 100원인 상품을 000원에 판매한 경우라면, 아래와 같이 -900 원 로그를 생성하어 kafka 에 추가하는 방식입니다.
// 원본 로그를 찾은 뒤
{"creativeNo":1004, "spend":1000, "impCount":0, "clickCount":1}
// -900 원으로 보정해서 통계 로그 토픽에 아래의 로그를 추가
{"creativeNo":1004, "spend":-900, "impCount":0, "clickCount":0}누적 데이터를 저장하는 mysql 에 저장하는 streaming 처리도 마이너스 처리를 반영하여, mysql 통계에도 동시에 반영이 가능합니다.
대규모 데이터 수정 — 세그먼트 삭제 후 재생성
기존에 없던 dimension, matric 추가 삭제나 보정으로는 해결이 안되는 대규모 수정의 경우 HDFS 의 파일을 hadoop mapreduce 를 통해 저장하는 hadoop ingestion 을 사용하였습니다.
데이터 백업 및 Hive, Spark 연동을 위해 HDFS 에 원본 로그와 통계 로그는 저장을 하고 있어서, 필요한 경우 HDFS 파일에 원하는 수정을 한 뒤 ingestion 을 수행하였습니다.
데이터 정합성
HDFS 에 저장된 통계 로그가 데이터 수정, 데이터 복구에서 사용이 되고, spot 데이터 추출에서도 사용이 되어 매시간/매일 배치를 통해 druid 와 hdfs 의 수치가 일치하는지 확인하고 있습니다.
hdfs 의 통계는 spark-sql 로 추출하고, druid 의 데이터를 쿼리를 통해 추출한 뒤 비교를 하여 문제가 있다면 알람을 받아 대응을 하고 있습니다.
다행히 kafka 장애나 hadoop 장애 상황외에는 druid 가 유실없이 데이터를 잘 저장하고 있어서, 정합성은 문제 없이 유지되고 있습니다.
kafka-indexing service 실행
위에서 설명한 내용에 대해 실제로 druid 에서 작업을 등록하고 수행하는 과정을 설명드리려 합니다.
druid 에서는 Rest Api 를 통해 작업을 등록 관리할 수 있고, 아래와 console gui 를 제공하여 작업 등록 수행/ 모니터링 쿼리 수행을 할 수 있습니다.
Load Data 메뉴를 통해서 Json 으로 작업(Datasource)을 등록할 수 있는데, 그 내용은 아래와 같습니다.
- actionTime 을 Time Dimension 으로 설정
- dimension 과 metric 모두 다른 이름으로 정의 가능
"timestampSpec": {
"column": "actionTime",
"format": "auto"
},
"dimensionsSpec": {
"dimensions": ["campaignNo", ... "advTimezone"]
}- metric 컬럼 설정
- Time Dimension 을 시간 단위 까지 조회하기 위해 queryGranularity 설정
{
"metricsSpec": [
{ "type": "longSum", "name": "impCount",
"fieldName": "impCount", "expression": null },
...
{ "type": "hyperUnique", "name": "uniqueUserCount",
"fieldName": "userId", "isInputHyperUnique": false, "round": false }
],
"granularitySpec": {
"type": "uniform",
"segmentGranularity": "DAY",
"queryGranularity": "HOUR",
"rollup": true,
"intervals": null
}
}등록한 Datasource 는 Supervisor 를 통해 관리가 되고 그 내용은 Ingestion 메뉴에서 확인하실 수 있습니다.
정상적으로 동작하는 데이터는 Query 메뉴에서 실제 쿼리를 수행하여 확인할 수 있습니다.
데이터 조회
Druid 에 저장한 데이터는 NativeQuery 나 Druid Sql 을 통해 조회할 수 있습니다.
Native Query
- Json object 기반의 쿼리로 데이터 조회에 관련 모든 기능 사용 가능
- Multi-dimension value 처리 가능
- Query cancellation. 가능
Druid Sql
- 친숙한 SELECT SQL 쿼리
- Rest Api, JDBC 지원
- Druid 내부적으론 Druid Sql 을 Native Query 로 치환하여 처리
- EXPLAIN PLAN FOR 로 PLAN 확인 가능
- useApproximateCountDistinct 을 사용할 수 있고, 기본 값이 true, COUNT(DISTINCT ) 를 사용하면 적용됨
- useApproximateTopN 사용가능, 기본값이 true 이고 GROUP BY, LIMIT 를 사용하면
Query 사용시 유의점
- time dimension 이 항상 Where 조건에 필요 (__time)
- timezone 이 default : UTC 로 동작함. gui 에서 timezone 을 세팅해 두거나 jdbc 의 경우 properties 에 , rest api 의 경우엔 header timezone 명시 가능
JDBC
- Apache Calcite Avatica 를 통해 JDBC 지원
- Spring jdbc template, Mybatis 사용 가능
- PreparedStatement 는 사용 불가
- JDBC GUI 툴을 통해 쿼리 가능
GFA 에서는 기존 SQL 의 익숙함과 기존 코드의 변경 최소화를 위해 Druid Sql 을 JDBC 방식을 통해 사용하였습니다.
데이터 조회 및 디버깅의 경우 datagrip 이나 dbeaver 에 druid connection 을 생성하여 쿼리를 하였습니다.
Distinct Count — useApproximateCountDistinct 사용
광고를 본 이용자의 숫자와 같은 unique count 계산을 위해useApproximateCountDistinct 를 적용하였습니다.
druid 에서 useApproximateCountDistinct 는 HyperLogLog ( 참고: D2 HyperLogLog) 를 이용하여 구현이 되어 있고, Druid SQL 로는 COUNT(DISTINCT) 를 이용하면 사용이 가능합니다.
bucket size 를 적용하는 옵션이 없어서 정확도를 조정할 수는 없지만, 실시간으로 아래과 같은 spec 설정으로 바로 사용이 가능합니다.
마무리
GFA 에서는 앞서 설명드린 것처럼 Druid 를 시계열 데이터 저장소로 선정하여 사용하고 있습니다.
아직 druid 0.x 버젼으라 많은 트러블 슈팅을 하고 있기는 하지만, 전체적으로 큰 문제 없이 잘 사용하고 있습니다.
Hadoop eco system 을 위에서 kafka 기반의 데이터를 시계열로 저장해야할 상황에서 고민 중이시라면 druid 를 테스트 후보로 올려 보아도 좋을 것 같습니다.
