[DEVIEW 리뷰] Luft: 10초만에 10억 데이터를 쿼리하는 데이터스토어 개발기

DEVIEW 2020 데이터 엔지니어링 파트의 Luft 세션을 들으면서 정리한 내용입니다.

40분 가량의 세션에서 기억하고 싶은 부분들이 너무 많아, 사실 거의 옮겨 적어놓은 것과 비슷합니다..ㅎㅎ

자세한 내용이 궁금하시다면 하단의 링크를 참고해주시고, 문제가 된다면 수정하겠습니다!

---

- 소개
Airbridge에서 자체 개발한 데이터스토어인 Luft의 개발기를 공유하고자 합니다.
Luft는 사용자 행동 분석에 최적화된 실시간 분산 데이터스토어입니다. S3과 Kafka로부터 각각 배치/실시간 데이터를 받아, 사용자 ID별로 파티셔닝해 저장하고 이를 바탕으로 리텐션 및 퍼널과 같은 사용자 행동 분석 쿼리를 수 초 이내로 수행합니다.
본 세션에서는 Luft에 대한 소개 및 개발기를 공유합니다. 국내에서 흔치 않은 데이터스토어 직접 개발이라는 경험을 통해, 대규모 분산처리 스택에 녹여져 있는 인사이트를 공유드리고자 합니다.

- 주제
이 발표를 통해, 이런 궁금증을 해결하실 수 있습니다.
왜 B2B 스타트업이 데이터스토어를 직접 개발했을까?대규모 고성능 애플리케이션을 만들 때 어떤 원칙이 필요할까?장애 허용되는 분산 시스템을 만들 때 무슨 어려움이 있고, 어떻게 해결할 수 있을까?우리가 평소에 쓰는 DB나 Spark, Hadoop 등이 어떤 생각의 흐름으로 만들어졌을까?

- 대상
이러한 내용에 대한 사전 지식이 있으신 데이터 엔지니어와 백엔드 개발자를 대상으로 합니다.
데이터베이스 샤딩 및 복제, Apache Spark/Hadoop, Data ETL / OLAP

출처: https://deview.kr/2020

 

1. 배경

1-1. Airbridge의 새로운 기능 개발

• 웹/앱의 유저 행동을 분석해 마케팅 성과를 측정하는 모바일 애널리틱스 서비스
• '코호트 분석: 커스텀 사용자군에 대한 분석'
  • 타겟 유저군을 자유자재로 -> 유저 ID에 대한 GROUP BY 쿼리를 자주, 빠르게 수행해야함, 시계열 분석
  • 분석 결과를 실시간으로 빠르게 제공해야 한다

1-2. 기존 솔루션의 한계

• Druid: 너무 느림. Pre-Aggregation 방식의 한계로 복잡한 행동분석에 한계
• Snowflake: SQL을 활용한 범용적인 데이터 웨어하우스이지만 느리고 비효율적

2. Luft

• 유저 행동 분석에 최적화된 실시간 OLAP 데이터스토어
• FAST, Real-Time, High Avaiability, Cloud Native

2-1. Architecture

• GO로 개발, gRPC로 통신
• Lambda Architecture
  • Master Node: API 서버 제공 및 쿼리 스케쥴링 & 결과 취합
  • Historical Node: 샤딩(batch 데이터 저장)된 데이터 저장 및 쿼리 연산
  • Realtime Node: 실시간 스트림 구독(Kafka) 및 쿼리 연산

 

2-2. Luft의 데이터 저장 방식

1. Luft의 데이터는 '사용자 행동 이벤트(구매, 장바구니 등)' 단위로 수집
2. 유저 ID를 기준으로 파티셔닝
3. 이벤트 발생 시각 기준으로 재파티셔닝 → 시계열 분석 쿼리에 최적화하기 위함!
4. 만들어진 파티션을 S3에 업로드, historical node에 분배됨 -> node가 파티션을 잃더라도 필요하면 S3에서 다시 불러올 수 있기 때문에 데이터 유실에 대한 걱정 없이 가용성을 유지할 수 있다.
5. 분배된 파티션은 리얼타임 노드의 데이터와 합쳐져 나중에 마스터 노드에 의해 쿼리됨

 

 

3. 개발기

• 기존 솔루션으로는 왜 문제를 해결하지 못하는지 충분히 이해하고, 오버엔지니어링을 피하자
  • → Druid, Spark, Kafka, CockroachDB 등 오픈소스 분석 및 인사이트 추출
    • ★Druid
      • Pre-Aggregation 방식. 정해진 metric을 시간 별로 미리 계산해두고, 쿼리 시엔 필요한 값만 reduce하여 빠르게 제공한다.
      • but, 정해진 종류의 한정된 분석만 가능 + High-Cardinal Shufflle에 취약
        • High-Cardinal Shuffle: 데이터를 정렬한 후, 다음 단계를 위해 네트워크를 통해 다시 모아야(reduce) 하는데, high-cardinality에 의해 데이터가 많이 퍼져있으면 그만큼 비용이 증가한다.
        • 유저 ID는 대표적인 high-cardinality column
        • 빠르게 자유로운 분석 결과를 제공하려는 개발 목적에 맞지 않음!
      • 해결방법: 유저 ID로 파티셔닝을 잘 해서 Shuffle이 국소적으로 일어날 수 있게 하자!
        • 데이터 저장 단계부터 미리 ID로 파티셔닝을 해두면 shuffle 단계의 부담이 적어질 것
• 기존 솔루션이 등장한 배경들과 지금은 무엇이 달라졌는지? 어떤 것을 추가로 이용할 수 있을지?
  • → CPU 코어 수, SIMD, 메모리 용량, I/O 속도, 클라우드
    • 클라우드를 최대한+효율적으로 사용함. 데이터는 S3, 메타데이터는 DynamoDB. k8s 워크로드 스케일링을 직접(..!) 관리하며 비용 절감

 

 

3-1. TrailDB에서 얻은 인사이트

• 마케팅 분석 솔루션에서 개발한 타임시리즈 이벤트 저장 rowstore.
• 유저 ID 기반(high-cardinality) 파티셔닝에 최적화된 유일한 스토리지 포맷
• 인덱스가 없지만 뛰어난 풀스캔 성능! 비결은,,
  • 충격적인 압축률(13GB → 300mb)과 O(N)의 시간 복잡도
    • 공간 복잡도를 극한으로 줄이면 낮은 시간 복잡도를 얻을 수 있다. 
    • 사이즈를 줄이면 메모리 활용 ↑, 거의 한달치의 데이터를 메모리에 올릴 수 있게 됨
  • 유저 이벤트 특성을 반영해 설계한 데이터 구조
    • 델타 인코딩: 유저의 이벤트를 시간순으로 정렬해, 이전 이벤트 대비 늘어난 시간값만 저장하여 효율적임
    • 딕셔너리 인코딩: 유저 ID를 제외한 데이터는 미리 사전화 시켜 ID만 저장
  • Columnstore같은 Rowstore
    • 엣지 인코딩: 이전 이벤트 대비 바뀐 칼럼만 저장. 대부분의 사용자 속성은 변하지 않기 때문!
    • Rowstore에서는 칼럼의 순서가 상관없기 때문에, 최빈순으로 정렬하여 Data Entropy를 줄임
• but, 수정이 불가능하고(하지만 HDFS, RDD 등 이미 일반적인 개념), 쿼리 엔진이나 샤딩, 복제, 클러스터링 같은 기능이 없음 (+데이터 타입이 String 밖에 없음)
  • 기존 RDBMS에서 강제되던 B+ Tree, skip list 같은 자료구조에서 벗어나고, 극한의 압축률을 구현할 수 있음
  • OLAP가 목적이기 때문에 수정 용이성을 버릴 수 있었던 것

 

3-2. LLVM으로 쿼리 엔진 개발

• TailDB 기본 기능의 문제점
  • Predicate Pushdown(스토리지 레벨에 필터를 적용해 필요한 데이터만 읽음) 기법은 필요하지만,
  • Go에서 파싱되는 쿼리를 C/C++로 넘겨야 함 + TrailDB에서 다양한 연산자를 지원하도록 쿼리 엔진 확장
  • Separation-of-Concern을 해침: 레이어화 과정에서 storage layer의 관심사를 application layer에서도 고려해아 하는 중복 문제 발생
• LLVM JiT
  • PostgreSQL에서 쿼리를 컴파일하기 위해 사용하는 컴파일러. IR 코드를 컴파일 할 수 있다.
  • Go에서 IR 코드 생성 →  C/C++에서 컴파일해 실행 가능
  • Separation-of-Concern 만족, 개발 비용 절감, 기능 추가 용이
• JiT 도입 과정에서 얻은 인사이트
  • JiT과 같은 복잡한 기술 스택을 도입하는 것은 일반적으로 오버엔지니어링
  • 다른 대안(직접 쿼리 엔진 개발)도 비슷하게 힘들기 때문에, 기능 확장이 용이한 쪽을 선택

3-3. 연산 레이어를 직접 만들기(LRMR)

• 보통 많이 쓰이는 MapReduce, Spark는 Go라서 못씀 + long-running job에 최적화되어 부적절함
• Less-Resilient MapReduce for Go, Go를 위한 MapReduce 오픈소스 프레임워크 개발
  • ★Spark의 디자인을 참고하여 golang과 gRPC로 MapReduce 연산 레이어 구현. 파티션 스케쥴링 방식과 Pull-based stream 구조를 통해 높은 성능으로 온라인 쿼리를 처리할 수 있었음.
  • Zookeeper 대신 etcd를 사용함(..!!). Zookeeper에 비해 운영도나 제약사항이 낮고 k8s에서 검증된 바 있음
  • Resiliency를 완전히 포기 -> 2013년 논문인 Piranha의 아이디어와 비슷
• 초기에는 Push-based stream 구조로 구현. 컨슈머의 속도가 프로듀서의 속도보다 느릴때 자주 나타나는 Backpressure 문제로 버퍼 오버플로가 자주 발생 → 컨슈머가 처리할 수 있는 양만 그때그때 요청하는 Pull-based stream 구조로 변경하여 해결

3-4. 샤딩 구현

• Luft에서의 샤딩: 파티션을 여러 히스토리컬 노드에 나누는 과정
• 파티션의 날짜 범위를 샤딩의 키 값으로 사용 → 데이터 필터링 & 분산 용이

x(시간)-y(클러스터 내의 노드)

구현 과정

1. 기본적인 Round-Robbin 알고리즘으로 새로운 파티션을 샤드에 분배
   • 분산 환경의 프로덕션에 적용하기에는 변수가 너무 많음
   • Rebalance, Time Decay, Eviction 등 고도화된 알고리즘 필요
2. Cost Function 사용
   • 스케쥴러에서 많이 사용하는 방식이지만 파티션을 분배하는 과정도 비슷하기 때문에 고려함
   • 파티션을 고르게 분배하기 위해서는, 두 파티션의 날짜 범위가 겹칠수록 코스트를 높게 하면 됨
   • 직접 구현하지는 못하고 Druid의 Cost Function을 베이스로 사용
3. 샤드의 가용성 확보
   • etcd로 장애상황을 효과적으로 관리. TTL 갱신을 사용한 Liveness Probe 패턴.
   • 클라우드를 적극적으로 활용

 

4. What's next?

• 현재는 실시간 코호트 모수 추출 및 리텐션 분석까지만 수행
• 실시간 퍼널 분석
• GROUP BY, JOIN 등 다양한 종류의 쿼리 지원
• 데이터 분석 확장성 및 ML 연동을 위한 Spark 지원
• 그리고.. 오픈소스화!
• Ziegel: TrailDB를 대체할 자체 Columnstore 개발.
  • TrailDB의 여러 한계점 해결 + 핵심 아이디어 계승
  • 데이터 기반 데이터스토어 설계
  • Bitmap Index 도입
  • SIMD와 멀티코어 최적화(병렬 처리, I/O 속도 향상)
• 실제 프로덕션 환경에서는 아직 경험이 짧음

5. 리뷰

실시간 쿼리를 처리하기 빠르게 처리하기 위해 데이터베이스 측면에서 문제를 어떻게 해결해나가는지,

여러 기술적인 대안들과 장단점 등에 대해 어떤 고민들을 해왔는지에 대한 과정들을 살펴볼 수 있어서 정말 좋았습니다.

문제 해결을 위한 생각의 흐름을 따라갈 수 있도록 세션이 구성되어 있어서 정말 재밌었던 것 같습니다!!

 

Short job을 MapReduce 측면에서(Hadoop의 Middelware로) 최적화하기 위한 논문을 공부한 적이 있는데

아이디어 중 하나는 check-pointing을 하지 않고, job이 실패하면 재시작한다는 것이었습니다.

같은 아이디어가 Luft 프로젝트의 연산 레이어를 개발하는 과정에서도 사용되어 왠지 반가운(..?) 느낌이 들었습니다!!

 

논문을 통해 공부했던 지식이 현업에서 문제 해결을 위해서도 사용되는 것을 보니까

그동안 공부했던 것들이 무의미하지 않구나라는 생각을 하게되었습니다.

이 개념이 보편적인 개념인지는 조금 궁금해지네요.

 

---

<출처>

DEVIEW 2020

Luft:10초만에 10억 데이터를 쿼리하는 데이터스토어 개발기

Luft: 유저 행동 분석에 최적화된 OLAP 데이터베이스