Dremel: Interactive Analysis of Web-Scale Datasets

1. INTRODUCTION

              ┌── Dremel ─────── Interactive analysis
              │
MapReduce ─── GFS ── Shared columnar data
              │
              ├── FlumeJava ─── Complex analysis
              │
              └── Pipeline ──── Production

• Dremel: read-only nested data를 분석하기 위한 확장 가능하고 대화형인 애드혹 쿼리 시스템
  • 다단계 실행 트리
  • column-oriented data batch
  • in situ에 있는 중첩 데이터를 직접 처리 → GFS같은 분산 파일 시스템이나 bigtable같은 다른 저장 계층에 놓인 데이터에 그대로 접근할 수 있음
• 데이터 응답 시간 → 데이터 탐색 / 모니터링 / 온라인 고객 지원 / 빠른 프로토타이핑 / 데이터 파이프라인 디버깅
• 병렬성
  • 한 클러스터에는 자원을 공유하는 여러 분산 애플리케이션이 함꼐 배치됨
  • 각각의 워크로드와 하드웨어 사양도 다름
  • 분산 애플리케이션의 어느 한 워커는 같은 작업을 실행하는 다른 워커보다 훨씬 오래 걸릴 수 있고, 장애나 클러스터 관리 시스템의 선점 때문에 완료하지 못할 수도 있음
  • → 빠른 실행과 내결함성을 확보하려면 straggler와 장애 처리가 필수적
• 유연한 데이터 모델이 중요
  • 정규화했다가 다시 결합하는 비용이 너무 비쌈

1. Serving tree
   • 분산 검색 엔진에서 사용
   • 쿼리는 트리 아래로 전달되며 각 단계에서 재작성 → 하위 단계에서 받은 응답을 집게해 최종 결과를 조립함
2. Ad-hoc 쿼리를 표현하기 위한 고수준의 SQL 유사 언어를 제공
   • unlike Pig/Hive → 쿼리를 MR 작업으로 번역하지 않고 직접 실행
3. Column-striped 저장 표현 사용
   • 보조 저장 장치에서 읽는 데이터 양 감소
   • 압축 비용이 저렴해짐

2. BACKGROUND

• e.g. 웹 페이지에서 새로운 종류의 신호를 추출
• 입력 데이터를 처리하는 MR 작업 → 해당 결과로 얻은 새로운 신호 수십억 개를 레코드 형태로 분산 파일 시스템에 저장 → 실험 결과를 분석하기 위해 다음 대화형 명령을 수행

DEFINE TABLE t AS /path/to/data/*
SELECT TOP(signal1, 100), COUNT(*) FROM t

flowchart LR

    A["Web Pages"]
    B["MapReduce<br/>Signal Extraction"]
    C[("Distributed File System<br/>(GFS)")]
    D["Dremel<br/>Interactive Analysis"]
    E["FlumeJava<br/>Deep Analysis"]
    F["Continuous<br/>Pipeline"]
    G["SQL<br/>Aggregation"]
    H["Interactive<br/>Dashboard"]
    I["Dataset<br/>Catalog"]

    A --> B
    B --> C
    C --> D

    D -->|"Irregularity found"| E
    E -->|"Fix & validate"| F

    F --> G
    G --> H
    F --> I

    C -.-> J["Shared Storage Layer"]
    J --> K["Shared Columnar Format"]
    K --> L["Read Only<br/>Required Fields"]

3. DATA MODEL

• Protocol Buffers
• 흠…

⭐ 4. NESTED COLUMNAR STORAGE

• goal: 검색 효율을 높이기 위해 특정 필드의 모든 값을 연속해서 저장하는 것

4.1. Repetition and Definition Levels

• 값만으로는 레코드 구조를 알 수 없음
  • 반복 필드의 값 두 개가 주어져도 어느 수준에서 값이 반복되었는지 = 서로 다른 두 레코드에서 온 것인지, 같은 레코드 안의 반복값 두 개인지
  • 선택 필드가 누락되었을 때 무엇이 명시적으로 정의되어 있었는지 알 수 없음.
• → Repetition Level / Definition Level

Document
├─ DocId                     required
├─ Links                     optional
│  ├─ Backward               repeated
│  └─ Forward                repeated
└─ Name                      repeated
   ├─ Language               repeated
   │  ├─ Code                required
   │  └─ Country             optional
   └─ Url                    optional

1. Repetition level

• field의 path에 있는 repeated field 중 어디에서 값이 반복되었는지
• e.g.

Name        repeated
 └ Language    repeated
     └ Code        required

• repetition level: 0~2
• r1

Name 1
 ├ Language
 │   └ Code = "en-us"
 ├ Language
 │   └ Code = "en"

Name 2
 └ Language 없음

Name 3
 └ Language
     └ Code = "en-gb"

value      repetition level

en-us             0 -> 새로운 record 시작
en                2 -> Langauage에서 반복 발생, r=2
NULL              1
en-gb             1 -> language가 아니라 새로운 Name으로 넘어감

• 문제: 두 번째 Name에는 Language 자체가 없으므로, code column을 단순히 en-us, en, en-gb라고 저장하면 en-gb가 Name2에 속한 건지 3에 속한 건지 구별하기 어려움.
• ⇒ Null을 넣음 = Name은 존재했으나 Code까지 내려가지는 않았다
• ⇒ 그러나 Name/Languge/Country중 어느 것이 없었는지 알기 어려움

2. Definition Level

• 해당 leaf field까지 가는 path에서 optional/repeqted field가 어디까지 실제로 존재했는지

Name          repeated
 └ Language      repeated
    └ Country       optional

• maximum definition level = 3
• is-null bit 대신 integer definition level을 사용
• ⇒ leaf column만 보더라도 parent field들이 실제로 존재했는지를 알 수 있음

value     r     d

us        0     3
NULL      2     2
NULL      1     1
gb        1     3
NULL      0     1

⇒ us, r=0, d=3 → 새로운 document, Name/Language/Country 모두 존재하고 Country는 “us”

이런 식으로 해석이 가능하다…

• NULL 값을 실제 value stream에 저장할 필요가 없음
  • ⇒ 당연함!! definition level을 보면 이미 NULL인지 알 수 있기 떄문
  • level도 작은 정수니까 bit packing 저장 가능

Nested records
      ↓
그냥 column으로 split하면 구조을 알 수 없음 -> 어디까지?
      ↓
r / d level로 구조만 아주 작게 따로 encoding
      ↓
전체 row를 읽을 필요 없이
필요한 leaf columns만 읽을 수 있음
      ↓
I/O 감소

4.2. Splitting Records into Columns

• r과 d를 실제로 어떻게 빠르게 계산해서 column을 만들지?!
• ⇒ 기본적으로 recode 구조를 재귀적으로 따라가면서 각 field value의 repetition/definition level을 계산
• problem) data는 sparse할 수 있음 = field가 수천 개 있더라도 실제 한 record에서는 전부 사용되지 않을 수도 있음.
• Field Writer Tree

              Document writer
             /               \
       Links writer        Name writer
        /      \            /       \
 Backward    Forward   Language     Url
                         /    \
                      Code    Country

• 이렇게 해놓고 각 leaf writer가 자기 column에 데이터를 씀
• 값이 없는 child까지 매번 업데이트하지 않음 / 부모의 level 상태를 inherit하고 있다가, 실제로 자기 값이 들어올 때만 부모와 동기화
• ~= lazy propagation

4.3. Record Assembly

• column들만 보고 원래 nested record를 어떻게 빠르게 조립하지?
• 4.2절과 반대
• FSM (=Finite State Machine)
  • 지금 column에서 값을 하나 읽었으면 다음에는 어느 column을 읽어야 하는지 FSM이 결정
  • 헐… 오토마타의 그것. 대충알겠당
  • 모든 field를 복원할 필요가 없음. 요청에 댑하기만 필요한 field만 사용하는 FSM을 만듦

5. QUERY LANGUAGE

• 가볍게 읽을 것!!
• Dremel의 언어는 SQL 기반인데 입출력은 둘 다 nested table임
• e.g.

SELECT
    DocId AS Id,
    COUNT(Name.Language.Code) WITHIN Name AS Cnt,
    Name.Url + ',' + Name.Language.Code AS Str
FROM t
WHERE REGEXP(Name.Url, '^http') AND DocId < 20;

• Path Expression
  • Name.Language.Code같은 nested path를 별도 join이나 flatten 없이 그대로 참조
  • 저장도 leaf path 단위로 했으니까 쿼리할 때도 그대로 처리
• WHERE
  • 조건을 만족하지 않는 nested branch를 잘라낸다(prune)
  • selection on a nested record = pruning branches of a labeled tree
• Projection
  • 어느 nesting level에서 값을 만들지
  • SELECT expression에서 가장 깊게 반복되는 input field와 같은 nesting level에서 output value를 만듦
• WITHIN
  • nested record 내부 aggregation
  • COUNT(Name.Language.Code) WITHIN Name
  • nested aggregation 결과도 nested structure를 유지

e.g.

Document
└─ Name 1
   ├─ Language
   │  └─ Code = en-us
   └─ Language
      └─ Code = en

이면

COUNT(Code) WITHIN Name
→ 2

다른 Name이

Name 2
└─ Language 없음

이라면

COUNT(Code) WITHIN Name
→ 0

결과는

Document
└─ Name
   └─ Cnt = 2

└─ Name
   └─ Cnt = 0

• 흠. 글쿤
• 이런것도 지원
  • nested subqueries
  • inter-record aggregation
  • intra-record aggregation
  • top-k
  • joins
  • user-defined functions

⭐ 6. QUERY EXECUTION

• 그 SQL을 수천 대의 서버가 어떻게 나눠서 몇 초 안에 실행하느냐?
• 쿼리를 한 서버가 처리하지 않고 multi-level serving tree로 내려보냄

Client
  │
  ▼
Root Server
  │
  ▼
Intermediate Servers
  │
  ▼
Leaf Servers
  │
  ▼
Storage Layer (e.g. GFS)

• Root: 쿼리를 받고 전체 작업을 나눔
• Intermediate: 하위 결과를 중간 집계
• Leaf: 실제 데이터를 읽고 scan/aggregation 수행
• Storage: columnar data가 저장돼 있음
• e.g.

SELECT A, COUNT(B)
FROM T
GROUP BY A

• Root server: T라는 table이 어떤 tablets로 구성되어 있지?!
• query rewriting: 각 서버가 자기 partition만 가지고 partial aggregation을 하게 함
• leaf가 실제 tablets을 병렬로 scan한 뒤 intermediate server가 partial result를 aggregate함 (→ root가 다 처리하면 완전 병목이 되겠지,,,)
• aggregation도 계층적으로 병렬화한다!!

• Columnar Storage → 적게 읽고
• Serving Tree → 빨리 나눠 읽고
• Query Dispatcher
  • multi-user system: 동시에 여러 쿼리가 들어옴
  • 그래서 query dispatcher가 뭘 하냐면,,,

1. Query scheduling
2. Priority handling
3. Load balancing
4. Straggler / failure handling

• Slot
  • 1 slot = leaf server 하나의 execution thread
  • e.g. 3000 leaf servers × 8 threads/server = 24,000 slots
  • 데이터를 tablet으로 쪼개놓고, 많은 slot이 동시에 처리한다
• straggler
  • 오래 걸리면 기다려야 함
  • dispatcher: 각 tablet 처리 시간을 histogram으로 모니터링
  • → 같은 tablet을 multiple times redispatch할 수 있음
  • tail latency 줄이기 위해서!
• Replication
  • tablet 데이터는 여러 replica를 가지고 있음(보통 3-way)
  • fault tolerance / query latency
• 최소 몇 %의 tablets를 scan한 뒤 결과를 반환할 것인지를 지정하는 parameter
  • 꼭 다 읽어야 하냐? → 그렇지 않을수도…
  • trade-off
  • approximate analysis라면 꽤 좋을수도?
• leaf에서는 column block을 prefetch함
  • asynchronously prefetch
  • I/O latency를 compute와 겹치는 개념? 인듯
• execution tree
  • physical query execution plan에 대응
  • scalar expression 처리
  • project-select-aggregate query: input columns를 lockstep으로 scan하면서 aggregate / scalar function 결과를 직접 만듦
  • record assembly를 아예 bypass할 수 있음
    • 흠…
  • 저장만 columnar한 게 아니라 execution까지 column-oriented하게 할 수 있다
• 그래서 왜 빠르냐면
  • Massive Parallel scan
  • Hierarchical aggregation
  • tail-latency mitigation
  • column-oriented execution

      Storage
         │
         ├─ Columnar layout
         │    → Read fewer bytes
         │
         ├─ r/d levels
         │    → Keep nested structure cheaply
         │
         ▼
      Execution
         │
         ├─ Thousands of leaf servers
         │    → Massive parallelism
         │
         ├─ Multi-level serving tree
         │    → Parallel aggregation
         │
         ├─ Direct column processing
         │    → Avoid record assembly
         │
         └─ Redispatch + replication
              → Reduce tail latency

7. EXPERIMENTS

• Columnar storage → 훨씬 적은 I/O
• Dremel execution → minutes → seconds
• 노드 수 증가 시 거의 선형적으로 빨라짐!! < 헐…
• Serving tree가 큰 aggregation에 효과적
• Straggler가 tail latency의 주요 원인
  • 실험을 했다 잘 했 다
  • 아무튼 빠름

8. OBSERVATIONS

• Trillion-row scan도 interactive speed 가능
• Columns / servers에 대해 near-linear scalability
• Columnar storage는 MR에도 유효
• Record assembly / parsing은 비싸다
• MR과 Dremel은 대체재가 아니라 complementary
• 마지막 몇 %의 straggler가 어렵다!!