.
분산 SQL 엔진 등장 이유
지난번에 다룬 SQL 엔진은 하나의 머신 안에서 메모리와 CPU를 활용해 쿼리를 실행하는 모델이었다. 하지만 데이터가 수십~수백 TB 단위로 커지면 단일 머신은 두 가지 한계에 부딫힌다.
•
메모리 한계: Hash Join의 build phase처럼 전체 데이터를 메모리에 올려야 하는 연산이 머신 한 대의 RAM을 초과
•
CPU 한계: 하나의 쿼리를 처리하는 데 걸리는 시간이 코어 수에 의해 제한됨
분산 SQL 엔진은 이 문제를 하나의 쿼리를 여러 머신에 쪼개서 동시에 실행하는 방식으로 해결한다. 핵심 아이디어는 operator 그래프 자체(Scan → Filter → Join → Aggregate)는 유지하면서 각 operator를 여러 노드에서 병렬로 실행할 수 있도록 데이터를 쪼개고(partition) 노드 간에 주고받는(shuffle) 구조를 추가하는 것이다.
단일 머신에서는 Scan 하고 Filter 하고 Join 한다 까지만 정해지면 됐는데, 분산 환경에서는 여기에 이 Scan을 몇개의 노드가 나눠서 할지, Join을 하려면 어떤 데이터를 어느 노드로 옮겨야 할지 같은 질문들이 추가된다. 즉 무엇을 할지(operator)는 같지만 어디서 몇개의 노드가 어떻게 나눠서 할지를 결정하는 단계가 새로 생긴다.
추가로 Trino, Presto 같은 엔진은 데이터베이스와 달리 저장소와 엔진이 분리되어 있다는 차이도 있다. 엔진은 S3, HDFS, 다른 DB 등 다양한 소스를 커넥터로 붙여 읽기만 하고 데이터를 자체 포맷으로 저장하지 않는다. 이 분리가 분산 실행 모델에도 영향을 준다 (어디서 데이터를 읽어오는지가 노드마다 다를 수 있음)
Stage: 분산 Physical Plan의 단위
단일 머신에서는 Physical Plan이 결정되면 그대로 operator 트리를 따라 실행하면 된다. 분산 엔진에서는 여기서 한 단계 더 나아가 Physical Plan을 Stage(또는 Fragment) 단위로 쪼갠다.
Stage 0: Scan(orders) → Filter
Stage 1: Scan(users)
Stage 2: Hash Join(Stage 0, Stage 1) → Aggregate
JavaScript
복사
Stage를 이해하려면 두 가지 개념을 구분해야 한다.
Stage — 무엇을 할지
하나의 Stage는 operator 묶음 하나를 의미한다. 위 예시에서 Stage 0은 orders를 읽고 필터링하는 일을 담당하고, Stage 2는 두 Stage의 결과를 받아 Join과 Aggregate를 수행한다.
Task — 누가 실행할지
하나의 Stage는 여러 노드에서 동시에 같은 연산을 수행한다. 예를 들어 Stage 0이 4개의 노드에 나뉘어 실행된다면 각 노드가 처리하는 작업 단위를 Task라고 부른다.
Stage 0 (4개 노드에서 병렬 실행):
Task 0-1: orders의 1/4를 Scan → Filter
Task 0-2: orders의 1/4를 Scan → Filter
Task 0-3: orders의 1/4를 Scan → Filter
Task 0-4: orders의 1/4를 Scan → Filter
JavaScript
복사
Stage는 무슨 작업인지를 정의하고 그 작업이 몇 개의 Task로 나뉘어 몇 개의 노드에서 동시에 도는지는 별도로 결정된다.
Stage 간 연결: 어디서 데이터를 주고받는가
Stage가 나뉘는 경계는 곧 노드 간 데이터 이동이 일어나는 지점이다. Scan → Filter처럼 같은 노드 안에서 끝낼 수 있는 연산은 하나의 Stage 안에 묶이지만, Join이나 GROUP BY처럼 여러 노드의 데이터를 모아야 하는 연산은 Stage를 나눠서 그 사이에 데이터를 전송하는 구간(shuffle)을 만든다.
Stage 0 (orders 처리) ──┐
├──→ Stage 2 (Join → Aggregate)
Stage 1 (users 처리) ──┘
JavaScript
복사
Stage 0과 Stage 1은 서로 독립적이라 동시에 실행될 수 있고 둘의 결과가 준비되면 Stage 2가 그걸 받아 Join을 수행한다. Stage 0, 1에서 Stage 2로 넘어가는 이 구간이 바로 데이터가 노드 간 네트워크를 타고 이동하는 지점이고 이 부분은 다음 섹션(Partitioning과 Shuffle)에서 자세히 다룬다.
정리하면, 분산 SQL 엔진의 실행 계획은 operator 트리를 그대로 쓰는 게 아니라 Stage 단위로 쪼개고 → 각 Stage를 여러 Task로 병렬화하고 → Stage 사이의 데이터 이동 지점을 정의하는 형태로 한 단계 더 구체화된다.
Partitioning과 Shuffle
Stage를 여러 노드에서 병렬로 돌리려면 데이터를 노드별로 어떻게 나눌지부터 정해야 한다. 이걸 결정하는 게 Partitioning이고 Stage 사이에서 데이터가 노드를 넘어 이동하는 과정이 Shuffle이다.
Partitioning 방식
데이터를 노드(또는 Task)에 나눠 담는 방법은 크게 세 가지가 있다.
방식 | 설명 |
Hash Partitioning | 특정 컬럼의 hash 값으로 노드를 결정 (Join key에 자주 사용) |
Range Partitioning | 값의 범위로 분배 (Sort에 유리) |
Round-Robin | key 무관 순서대로 균등 분배 (단순 분산) |
Shuffle이 발생하는 시점
Join이나 GROUP BY처럼 같은 key를 가진 row들이 같은 노드에 모여야 하는 연산에서 shuffle이 발생한다. 예를 들어 GROUP BY user_id를 여러 노드에서 병렬로 정확하게 계산하려면 같은 user_id를 가진 row들이 전부 같은 노드로 모여야 한다. 그래야 그 노드가 자신이 받은 데이터만으로 정확한 집계를 낼 수 있다.
Before Shuffle (노드별 무작위 분산):
Node1: user_id=1,2,1,3
Node2: user_id=2,1,3,2
After Shuffle (user_id 기준 hash partition):
Node1: user_id=1,1
Node2: user_id=2,2,2
Node3: user_id=3,3
JavaScript
복사
Shuffle 비용
Shuffle이 발생하면 모든 노드가 가진 데이터를 다른 노드로 보내거나 받아야 한다. 이 과정에서:
•
네트워크를 통해 데이터가 이동 (디스크 I/O보다 메모리 접근은 빠르지만, 네트워크는 그 둘보다 느림)
•
받는 쪽 노드는 데이터가 어느 정도 모일 때까지 대기해야 하는 경우가 많음 (Stage 경계 = materialization 지점)
•
데이터가 한쪽 노드에 몰리면(skew) 그 노드가 병목이 됨
이런 이유로 분산 엔진에서 Shuffle은 가장 비용이 큰 연산 중 하나로 취급된다. 실제 쿼리가 느릴 때 실행 계획을 열어보면, Shuffle이 몇 번 발생하는지 각 Shuffle에서 얼마나 많은 데이터가 이동하는지를 확인하는 게 성능 분석의 핵심 포인트가 된다. Join 순서를 바꾸거나 필터를 앞으로 당기는(Predicate Pushdown) 최적화도 결국 Shuffle이 일어나기 전에 데이터 양을 최대한 줄이려는 목적이 크다.
Stage와 Shuffle의 관계
앞서 본 Stage 구조를 다시 보면:
Stage 0 (orders 처리) ──┐
├──→ Stage 2 (Join → Aggregate)
Stage 1 (users 처리) ──┘
JavaScript
복사
Stage 0과 Stage 1의 결과가 Stage 2로 넘어가는 지점이 바로 Shuffle이 발생하는 곳이다. Stage 0이 orders를 join key 기준으로 hash partitioning해서 내보내고 Stage 2의 각 노드는 자기 partition에 해당하는 데이터만 받아서 처리한다. Stage 경계가 곧 데이터 이동 경계이고 그 이동 방식을 결정하는 게 partitioning이라고 보면 된다.
분산 Join: Broadcast vs Partitioned
Hash Join에는 작은 테이블을 메모리에 해시 테이블로 만들고 큰 테이블을 흘려보내며 매칭하는 build phase 단계가 있다. 분산 환경에서는 이 build phase를 모든 노드에서 어떻게 준비시킬지에 따라 두 가지 전략이 갈린다.
Broadcast Join
작은 테이블을 전체 복제해서 모든 노드에 그대로 전송하는 방식
orders (1억 건) JOIN users (1만 건)
users (작음) → 모든 노드에 통째로 복제
orders (큼) → 그대로 두고 각 노드가 자기 partition만 처리
JavaScript
복사
Node1: orders 1/4 + users 전체(복제본)
Node2: orders 1/4 + users 전체(복제본)
Node3: orders 1/4 + users 전체(복제본)
Node4: orders 1/4 + users 전체(복제본)
JavaScript
복사
•
큰 테이블(orders)은 그대로 두고 각 노드가 자기 partition만 처리
•
Shuffle이 큰 테이블 쪽에서는 발생하지 않고 작은 쪽만 복제 비용 발생
•
조건: 한쪽 테이블이 메모리에 들어갈 만큼 작아야 함
Partitioned (Shuffle) Join
양쪽 테이블이 모두 클 때는 복제할 수 없다. 대신 양쪽 다 join key 기준으로 hash partitioning해서 Shuffle한다.
•
같은 key를 가진 row들이 동일 노드에 모이도록 해시함수 적용
•
각 노드에서 로컬 Hash Join 수행
•
양쪽 다 Shuffle이 발생하는 대신 복제 비용은 없음
옵티마이저 판단 예시
orders (1억) JOIN users (1만)
옵티마이저 판단:
users의 예상 크기가 메모리 임계값 이하 → Broadcast Join 선택
모든 노드에 users 복제 → orders는 그대로 각자 처리
JavaScript
복사
대체로 통계 정보(row count, 평균 row 크기)로 추정한 테이블 크기로 판단한다. 여기서 통계가 부정확하면 문제가 생긴다. 실제로는 큰 테이블인데 옵티마이저가 작다고 잘못 추정해서 Broadcast를 선택하면 메모리가 터질 수 있다.
이런 문제는 통계를 주기적으로 갱신하거나 Join 전략을 직접 힌트로 강제하는 방식으로 대응한다.
Execution: Coordinator와 Worker
지금까지 본 Stage 분해, Partitioning, Join 전략 선택은 모두 누가 결정하고 누가 실행하는가라는 질문으로 이어진다. 분산 SQL 엔진은 대체로 이 역할을 두 가지로 나눈다.
Coordinator
쿼리를 받아서 실행 준비를 끝내는 노드
•
쿼리를 Parsing하고 Logical Plan, Physical Plan을 생성
•
Physical Plan을 Stage로 분해
•
각 Stage를 몇 개의 Task로 나눌지, 어느 Worker에 배정할지 스케줄링
•
Worker들의 진행 상태를 추적하고, 최종 결과를 모아 클라이언트에 반환
쿼리 하나당 Coordinator는 보통 하나만 동작한다. 클러스터 전체에서 이 쿼리를 어떻게 나눠서 돌릴지를 결정하는 단일 지점이다.
Worker
실제로 데이터를 읽고 연산을 수행하는 노드
•
Coordinator가 배정한 Task를 받아 실행 (Scan, Filter, Join, Aggregate 등 operator 수행)
•
필요하면 다른 Worker와 Shuffle을 통해 데이터를 주고받음
•
Task 완료 시 결과를 다음 Stage로 전달하거나 Coordinator에 보고
Worker는 클러스터에 여러 대가 있고 하나의 쿼리는 보통 여러 Worker에 걸쳐 동시에 실행된다.
전체 흐름
1. 클라이언트 → Coordinator에 SQL 전송
2. Coordinator: Parsing → Optimization → Physical Plan → Stage 분해
3. Coordinator: 각 Stage를 Task로 나눠 Worker들에 배정
4. Worker들: Task 실행, 필요시 서로 Shuffle
5. 마지막 Stage 결과 → Coordinator → 클라이언트
JavaScript
복사
Coordinator
├─ Stage 0 → Worker1, Worker2 (orders 처리)
├─ Stage 1 → Worker3 (users 처리, Broadcast 대상이면 모든 Worker에 복제)
└─ Stage 2 → Worker1, Worker2, Worker3 (Join → Aggregate)
JavaScript
복사
Stage 0과 Stage 1은 서로 의존성이 없으니 동시에 실행되고 그 결과가 준비되면 Stage 2가 받아서 처리한다. 이 의존 관계를 추적하고 스케줄링하는 게 Coordinator의 역할이다.
장애와 재시도
분산 환경에서는 노드 하나가 죽거나 느려지는 일이 흔하다. Coordinator는 Worker의 상태를 계속 모니터링하면서
•
특정 Task가 실패하면 다른 Worker에 재시도 배정
•
특정 Worker가 유독 느리면(straggler) 같은 Task를 다른 Worker에도 중복 실행해서 먼저 끝나는 쪽 결과를 사용하는 전략을 쓰기도 함
단일 머신 엔진에서는 이 머신이 죽으면 쿼리도 끝이지만 분산 엔진은 일부 Worker 장애를 감내하고 쿼리를 끝까지 끌고 가야 하는 책임이 추가된다.
정리
SQL → Logical Plan → Physical Plan → Stage 분해 → Partitioning/Shuffle 결정 → Coordinator가 Task 스케줄링 → Worker들이 병렬 실행 → 결과 수집
단일 머신 엔진의 컴파일-실행 구조 위에 이 작업을 몇 개의 노드가, 어떤 방식으로 데이터를 나눠 병렬로 처리할 것인가라는 질문과, 그걸 누가 결정하고(Coordinator) 누가 수행하는지(Worker)를 정의하는 구조가 한 겹 더 추가된 것이 분산 SQL 엔진이다.