Hive의 비용 기반 최적화

Hive의 비용 기반 최적화 (Cost-based optimization in Hive)

Apache Hive는 사용자가 제출한 SQL 쿼리를 물리 연산자 트리로 변환하고, 이를 Tez 잡으로 변환한 뒤 Hadoop 클러스터에서 실행해요. 기존 Hive 최적화는 대부분 셔플링 비용 최소화에 집중했는데, Apache Calcite 기반의 비용 기반 최적화기(CBO)를 도입하면 조인 순서와 조인 알고리즘을 자동으로 선택해 쿼리 지연 시간을 크게 줄일 수 있답니다.

출처: 문서

본문

초록 (Abstract)

Apache Hadoop은 일반적으로 상용 하드웨어로 구성된 컴퓨터 클러스터를 사용해 대규모 데이터셋을 분산 처리하는 프레임워크예요. 지난 몇 년간 Apache Hadoop은 상용 하드웨어를 사용한 분산 데이터 처리의 사실상 플랫폼이 됐어요. Apache Hive는 Apache Hadoop을 사용한 데이터 처리의 인기 있는 SQL 인터페이스예요.

사용자가 제출한 SQL 쿼리는 Hive가 물리 연산자 트리로 변환하고, 이 트리는 최적화된 뒤 Tez 잡으로 변환돼 Hadoop 클러스터에서 실행돼요. Hadoop에서의 분산 SQL 쿼리 처리는 중간 결과 집합을 다룰 때 기존 관계형 쿼리 엔진과 달라요. Hive 쿼리 처리는 종종 중간 결과 집합의 정렬과 재조립을 요구하는데, 이를 Hadoop 용어로 셔플링(shuffling)이라고 해요.

Hive의 기존 쿼리 최적화 대부분은 셔플링 비용 최소화에 관한 것이에요. 현재 사용자는 쿼리가 효율적으로 실행되도록 올바른 조인 순서로 최적화된 쿼리를 Hive에 제출해야 해요. Hive의 논리적 최적화는 필터 푸시다운, 프로젝션 프루닝, 파티션 프루닝으로 제한돼요. 비용 기반 논리적 최적화는 Apache Hive의 쿼리 지연 시간과 사용 편의성을 크게 개선할 수 있어요.

조인 재정렬(join reordering)과 조인 알고리즘 선택은 비용 기반 최적화기의 혜택을 받을 수 있는 최적화 중 일부예요. 비용 기반 최적화기는 사용자가 조인을 올바른 순서로 재배열하거나 쿼리 힌트와 구성 옵션으로 조인 알고리즘을 지정해야 하는 부담에서 해방시켜줘요. 이는 잠재적으로 사용자가 쿼리 최적화에 신경 쓰지 않고 비즈니스 프로세스에 가깝게 보고(reporting)와 ETL 요구를 모델링할 수 있게 해줘요.

Calcite는 오픈 소스 비용 기반 쿼리 최적화기이자 쿼리 실행 프레임워크예요. Calcite는 현재 쿼리 트리를 재작성할 수 있는 50개 이상의 쿼리 최적화 규칙과, 가장 저렴한 쿼리 플랜을 최적으로 선택할 수 있는 효율적인 플랜 프루너를 보유하고 있어요. 이 문서에서는 Calcite를 사용해 Apache Hive에 비용 기반 논리적 최적화기(CBO)를 도입하는 방법을 논의해요.

CBO는 단계적으로 Hive에 도입될 거예요. 첫 번째 단계에서 Calcite는 조인을 재정렬하고 올바른 조인 알고리즘을 골라 쿼리 지연 시간을 줄이는 데 사용될 거예요. 이를 위해 테이블 카디널리티와 경계(boundary) 통계가 사용돼요.

1. 소개 (INTRODUCTION)

Hive는 Apache Hadoop 위의 데이터 웨어하우징 인프라예요. Hive는 Hadoop의 대규모 스케일 아웃과 내결함성 기능을 활용해 상용 하드웨어에서 데이터 저장과 처리를 해요. Hive는 쉬운 데이터 요약, 임시(ad-hoc) 쿼리, 대용량 데이터 분석을 가능하게 설계됐어요. Hive SQL은 선언적 쿼리 언어로, SQL에 익숙한 사용자가 임시 쿼리, 요약, 데이터 분석을 쉽게 할 수 있게 해줘요.

과거에는 Hadoop 잡의 지연 시간이 높았고 잡 제출·스케줄링에 상당한 오버헤드가 있었어요. 결과적으로 관련된 데이터셋이 매우 작더라도 Hive 쿼리의 지연 시간은 일반적으로 매우 높았어요. 그래서 Hive는 주로 ETL에 사용됐고 대화형 쿼리에는 많이 사용되지 않았어요. Hadoop2와 Tez로 잡 제출·잡 스케줄링의 오버헤드가 크게 줄었어요. Hadoop 1에서는 실행할 수 있는 잡이 Map-Reduce 잡뿐이었어요. Hadoop2와 Tez에서는 그 제한이 더 이상 적용되지 않아요.

Hadoop에서 매퍼의 출력은 정렬되고 때로는 매퍼의 로컬 디스크에 유지돼요. 이 정렬된 매퍼 출력은 적절한 리듀서로 보내지고, 리듀서는 서로 다른 매퍼의 정렬된 결과를 결합해요. 한 잡의 출력을 다음 map-reduce 잡이 소비해야 하는 여러 map-reduce 잡을 실행할 때, 이전 map-reduce 잡의 출력은 HDFS에 유지되어야 해요. 이 유지는 HDFS의 복제 계수에 따라 데이터를 다른 노드로 복사해야 하므로 비용이 커요.

Hadoop 1 위의 Hive는 쿼리 처리를 완료하기 위해 여러 map-reduce 잡을 제출해야 하는 경우가 많았어요. 이 map-reduce 잡 파이프라인은 중간 결과 집합을 이제 내결함성 HDFS에 유지해야 하므로 성능을 저하시켰어요. 또한 잡 제출과 스케줄링도 비교적 비싼 연산이었어요. Hadoop2와 Tez에서는 잡 제출·스케줄링 비용이 최소화됐어요. 또한 Tez는 잡을 Map 다음 Reduce로만 제한하지 않아서, 모든 쿼리 실행을 잡 경계를 넘지 않고 단일 잡으로 수행할 수 있어요. 이는 중간 결과 집합을 HDFS나 심지어 로컬 디스크에도 유지할 필요가 없으므로 상당한 비용 절감을 가져와요.

관계형 쿼리 엔진의 쿼리 최적화는 크게 논리적 쿼리 최적화와 물리적 쿼리 최적화로 분류할 수 있어요. 논리적 쿼리 최적화는 일반적으로 쿼리가 실행되는 물리적 계층과 무관하게 관계 대수에서 파생될 수 있는 쿼리 최적화를 말해요. 물리적 쿼리 최적화는 물리적 계층 프리미티브를 알고 있는 쿼리 최적화예요. Hive의 경우 물리적 계층은 Map-Reduce와 Tez 프리미티브를 의미해요.

현재 Hive의 논리적 쿼리 최적화는 크게 다음과 같이 분류할 수 있어요:

  • Projection Pruning
  • Transitive Predicates 유도
  • Predicate Push down
  • Select-Select, Filter-Filter를 단일 연산자로 병합
  • Multi-way Join
  • 일부 컬럼 값의 조인 스큐를 수용하기 위한 Query Rewrite

Hive의 물리적 최적화는 크게 다음과 같이 분류할 수 있어요:

  • Partition Pruning
  • 파티션과 버킷팅 기반 스캔 프루닝
  • 샘플링 기반 쿼리인 경우 스캔 프루닝
  • 어떤 경우 Map 측에서 Group By 적용
  • 매퍼에서 Join 수행
  • 유니온이 Map 측에서만 수행되도록 유니온 최적화
  • 멀티웨이 조인에서 사용자 힌트를 기반으로 어떤 테이블을 마지막으로 스트림할지 결정
  • 불필요한 reduce sink 연산자 제거
  • limit 절이 있는 쿼리에서 테이블을 스캔해야 하는 파일 수 줄이기
  • limit 절이 있는 쿼리에서 Reduce Sink 연산자가 생성하는 것을 제한해 매퍼 출력 제한
  • 사용자가 제출한 SQL 쿼리를 처리하는 데 필요한 Tez 잡 수 줄이기
  • 단순 fetch 쿼리의 경우 Map-Reduce 잡 회피
  • 집계가 있는 단순 fetch 쿼리에서 Map-Reduce 태스크 없이 집계 수행
  • Group By 쿼리를 원본 테이블 대신 인덱스 테이블을 사용하도록 재작성
  • 테이블 스캔 위의 술어가 동등(equality) 술어이고 술어의 컬럼에 인덱스가 있을 때 인덱스 스캔 사용

Hive에서 최적화 대부분은 쿼리 실행 비용에 기반하지 않아요. 대부분의 최적화는 필터 푸시다운과 연산자 병합을 제외하면 연산자 트리를 재배열하지 않아요. 연산자 트리 변경의 대부분은 reduce-sink와 reducer 연산자 제거를 위한 것d이에요. 아래는 CBO의 혜택을 받을 수 있는 몇 가지 최적화 결정이에요:

  • Join을 어떻게 정렬할 것인가
  • 주어진 Join에 어떤 알고리즘을 사용할 것인가
  • 중간 결과를 유지할 것인지, 연산자 실패 시 재계산할 것인지
  • 어떤 연산자에서의 병렬도 (특히 사용할 리듀서 수)
  • Semi Join 선택

Calcite는 오픈 소스이며 Apache License를 사용하는 쿼리 계획·실행 프레임워크예요. Calcite의 많은 부분은 Eigenbase 프로젝트에서 파생됐어요. Calcite는 선택적 JDBC 서버, 쿼리 파서와 검증기, 쿼리 최적화기, 플러그형 데이터 소스 어댑터를 갖고 있어요. 사용 가능한 Calcite 최적화기 중 하나는 volcano 페이퍼에 기반한 비용 기반 최적화기예요. 현재 Calcite의 여러 부분은 다음 프로젝트/제품에서 사용돼요:

  • Apache Drill
  • Cascading (Lingual)
  • Lucid DB
  • Mondrian/Pentaho

Calcite는 현재 50개 이상의 비용 기반 최적화 규칙을 갖고 있어요. 기존 비용 기반 최적화 규칙 중 몇 가지 주요 규칙은 다음과 같아요:

  • Push Join through Union
  • Push Filter past Table Function
  • Join Reordering
  • Semi Join selection
  • Push Aggregate through Union
  • Pull Aggregate through Union
  • Pull Constants through Aggregate
  • Merge Unions

이 문서에서는 Hive에서 비용 기반 최적화를 수행하기 위해 Calcite의 비용 기반 최적화기인 Volcano를 사용할 것을 제안해요. Calcite 기반 CBO를 단계적으로 구현할 것을 제안해요. 여기서 제안은 Calcite의 최적화기만 사용하고 다른 것은 사용하지 않는 것에요. Calcite를 사용해 Hive에 CBO를 도입하는 구상 단계는 다음과 같아요:

  • Phase1 - Join Reordering & Join algorithm Selection
    • 테이블 카디널리티와 경계 통계로 연산자 카디널리티를 계산.
    • Hive 연산자 트리를 Calcite 연산자 트리로 변환.
    • Calcite의 Volcano 최적화기로 조인을 재배열하고 조인 알고리즘을 선택.
    • 최적화된 Calcite 연산자 트리를 다시 Hive AST로 변환해 이전처럼 실행. 따라서 Hive의 모든 기존 최적화는 Calcite 최적화 SQL 위에서 실행됨.
  • Phase2 - 히스토그램 지원 추가, Calcite의 다른 최적화 사용
    • 공간 효율적인 히스토그램 도입
    • 연산자 카디널리티 계산을 히스토그램 사용으로 변경
    • 위에 나열된 것 같은 Calcite의 추가 최적화 규칙 등록
  • Phase3 - Calcite가 최적화된 Hive 연산자 트리를 생성하도록 코드 재구성
    • phase1과 달리 Hive AST가 Calcite 연산자 트리로 직접 변환됨.
    • Volcano 최적화기로 Calcite 연산자 트리 최적화.
    • 최적화된 Calcite 연산자 트리를 다시 Hive 연산자 트리로 변환. 이는 최적화된 Calcite 연산자 트리를 Hive AST로 변환하는 phase1과 다름.

STATS

PAPERS

  • Query Optimization for Massively Parallel Data Processing
    • Sai Wu, Feng Li, Sharad Mehrotra, Beng Chin Ooi
    • School of Computing, National University of Singapore, Singapore, 117590
    • School of Information and Computer Science, University of California at Irvine
  • Profiling, What-if Analysis, and Cost-based Optimization of MapReduce Programs
    • Herodotos Herodotou Duke University, Shivnath Babu Duke University
  • Optimizing Joins in a Map-Reduce Environment
    • Foto N. Afrati, National Technical University of Athens, Greece / Jeffrey D. Ullman, Stanford University USA
  • Efficient and scalable statistics gathering for large databases in Oracle 11g
  • Estimating Distinct (Postgress SQL)
  • The History of Histograms

3. 배경 (BACKGROUND)

Hive 쿼리 최적화 문제점

Hive는 사용자가 지정한 SQL 문을 AST로 변환하고, 그 AST로 물리 연산자 트리를 만든다. 모든 쿼리 최적화는 물리 연산자 트리에서 수행돼요. Hive는 의미(semantic) 정보를 쿼리 연산자 트리와 분리해 유지해요. 의미 정보는 플랜 생성 중 추출되며, 이후 다운스트림 쿼리 최적화에서 자주 조회돼요. 적절한 논리적 쿼리 플랜 부족과 의미 정보·쿼리 트리 분리 때문에 Hive에 새 쿼리 최적화를 추가하는 것이 종종 어려워져요.

TEZ

Apache Tez는 복잡한 태스크 DAG(directed acyclic graph)를 실행하기 위해 MapReduce 패러다임을 일반화해요. 자세한 내용은 다음 링크를 참고해요.

http://hortonworks.com/blog/apache-tez-a-new-chapter-in-hadoop-data-processing/

Hive의 조인 알고리즘

Hive는 현재 equi-Join만 지원해요. Hive 조인 알고리즘은 다음 중 하나가 될 수 있어요.

Multi way Join

여러 조인이 같은 드라이빙 측 조인 키를 공유하면 모든 조인을 단일 태스크에서 할 수 있어요.

예: (R1 PR1.x=R2.a - R2) PR1.x=R3.b - R3) PR1.x=R4.c - R4

모든 조인은 같은 리듀서에서 수행될 수 있는데, R1이 조인 키 x로 이미 정렬되기 때문이에요.

Common Join

조인 키로 테이블의 병렬 정렬을 수행하는 데 Mapper를 사용하고, 그 결과를 리듀서로 넘겨요. 같은 키를 가진 모든 튜플이 같은 리듀서에 주어져요. 리듀서는 한 개 이상의 키에 대한 튜플을 받을 수 있어요. 튜플의 키에는 테이블 id도 포함되므로, 같은 키를 가진 두 다른 테이블의 정렬된 출력을 구분할 수 있어요. 리듀서는 정렬된 스트림을 병합해 조인 출력을 얻어요.

Map Join

스타 스키마 조인에 유용한 이 조인 알고리즘은 모든 작은 테이블(차원 테이블)을 모든 매퍼의 메모리에 유지하고 큰 테이블(팩트 테이블)은 매퍼에서 그 위로 스트리밍해요. 이는 Common-Join에 내재된 셔플링 비용을 피해요. 각 작은 테이블(차원 테이블)에 대해 조인 키를 해시 테이블 키로 사용해 해시 테이블을 만든다.

Bucket Map Join

map-join의 조인 키가 버킷화되어 있으면, 모든 매퍼에 작은 테이블(차원 테이블) 전체를 유지하는 대신 일치하는 버킷만 유지해요. 이는 map-join의 메모리 사용량을 줄여요.

SMB Join

이는 Bucket Map Join의 최적화예요. 조인할 데이터가 조인 키로 이미 정렬되어 있으면 해시 테이블 생성을 피하고 대신 정렬 병합 조인(sort merge join) 알고리즘을 사용해요.

Skew Join

일부 특정 값에 대해 데이터 분포가 기울어져(skewed) 있으면, 조인 연산자의 일부 인스턴스(map-reduce 세계의 리듀서)는 과부하되고 다른 인스턴스는 활용이 부족해져 조인 성능이 저하될 수 있어요. 사용자 힌트가 있으면 Hive는 스큐 값 주변의 조인 쿼리를 조인의 유니온으로 재작성해요.

예 R1 PR1.x=R2.a - R2에서 대부분의 데이터가 x=1 주변으로 분포하면, 이 조인은 (R1 PR1.x=R2.a and PR1.x=1 - R2) union all (R1 PR1.x=R2.a and PR1.x<>1 - R2)로 재작성될 수 있어요.

4. 구현 세부 사항 (Implementation details)

CBO는 세 가지 다른 단계로 Hive에 도입될 거예요. 다음은 이 단계들의 개략적인 개요예요.

Phase 1

Statistics (통계):

  • Table Cardinality
  • Column Boundary Stats: min, max, avg, number of distinct values

Cost Based Optimizations (비용 기반 최적화):

  • Join ordering
  • Join Algorithm

Restrictions (제약):

  • Calcite CBO는 select 표현식에만 사용됨
  • select 표현식에 다음 연산자 중 하나가 포함되면 Calcite CBO를 사용하지 않음:
    • Sort By

Hive는 전체 정렬(order by)과 부분 정렬(sort by)을 모두 지원해요. 부분 정렬은 관계 대수와 SQL로 표현할 수 없어요. 향후 Sort By를 테이블 함수로 표현할 수도 있어요.

  • Map/Reduce/Transform

Hive는 SQL에서 map/reduce/transform 연산자를 지정할 수 있게 해줘요. 제공된 매퍼/리듀서 스크립트로 데이터가 변환돼요. 이러한 연산자는 관계 대수로 직접 변환할 수 없어요. 향후 테이블 함수로 표현할 수 있어요.

  • Cluster By/Distribute By

Cluster By와 Distribute By는 주로 Transform/Map-Reduce Scripts와 함께 사용돼요. 하지만 후속 쿼리를 위해 쿼리 출력을 파티셔닝하고 정렬할 필요가 있을 때 SELECT 문에서 유용할 수 있어요. Cluster By는 Distribute By와 Sort By 둘 다의 축약형이에요. Hive는 Distribute By의 컬럼을 사용해 리듀서들 사이에 행을 분배해요. 같은 Distribute By 컬럼을 가진 모든 행은 같은 리듀서로 갑니다. 그러나 Distribute By는 분산된 키에 대한 클러스터링이나 정렬 속성을 보장하지 않아요.

  • Table Sample

TABLESAMPLE 절은 사용자가 전체 테이블 대신 데이터의 샘플에 대한 쿼리를 작성할 수 있게 해줘요. TABLESAMPLE 절은 FROM 절의 어떤 테이블에도 추가할 수 있어요. 향후 Table Sample을 테이블 함수로 표현할 수 있어요.

  • Lateral Views

Lateral view는 explode() 같은 사용자 정의 테이블 생성 함수(UDTF)와 함께 사용돼요. UDTF는 각 입력 행에 대해 하나 이상의 출력 행을 생성해요. lateral view는 먼저 기본 테이블의 각 행에 UDTF를 적용한 뒤 결과 출력 행을 입력 행과 조인해 지정된 테이블 별칭을 갖는 가상 테이블을 만들어요.

  • UDTF (Table Functions)
  • PTF (Partitioned Table Functions)

Calcite 관련 개선:

  • hive 관계형 연산자들을 표현할 연산자 도입. Table Scan, Join, Union, Select, Filter, Group By, Distinct, Order By. 이 연산자들은 각 연산자의 물리적 비용을 포함한 호출 규약을 구현.
  • Join을 CommonJoin에서 MapJoin, MapJoin에서 BucketJoin, BucketJoin에서 SMBJoin, CommonJoin에서 SkewJoin으로 변환하는 규칙 도입.
  • 단일 조인 연산자가 multi-way join을 나타내도록 조인을 병합하는 규칙 도입 (Hive의 MergedJoin과 유사).
  • Hive의 Merged-Join은 Calcite에서 MultiJoinRel로 번역됨.

Phase 2

Statistics (통계):

  • Histograms

Cost Based Optimizations (비용 기반 최적화):

  • 히스토그램 기반 Join ordering
  • Join Algorithm - 조인 선택도를 추정하는 데 히스토그램 사용
  • Calcite의 추가 최적화 활용. 실제 사용할 규칙은 TBD.

Phase 3

(단계별 이미지)

설정 (Configuration)

구성 매개변수 hive.cbo.enable는 비용 기반 최적화를 활성화할지 여부를 결정해요.

제안된 비용 모델 (Proposed Cost Model)

Hive는 Hadoop 클러스터를 사용해 병렬 분산 쿼리 실행을 사용해요. 이는 주어진 쿼리 연산자 트리에서 다른 연산자들이 다른 노드에서 실행될 수 있다는 것을 의미해요. 또한 같은 연산자가 클러스터의 다른 노드에서 병렬로 실행되어 원래 관계의 다른 파티션을 처리할 수 있어요. 이 병렬 실행 모델은 높은 I/O와 CPU 비용을 유발해요. Hive 쿼리 실행 비용은 다음 이유로 I/O 중심이 되는 경향이 있어요.

  • Shuffling cost

쿼리 트리에서 연산자가 자식 연산자로부터 필요한 데이터는 자식 연산자의 모든 인스턴스에서 데이터를 조립해야 해요. 이 데이터는 정렬되고 잘게 쪼개져 연산자의 인스턴스에 관계의 파티션이 제시돼요.

이 셔플링 비용은 중간 결과 집합을 로컬 파일 시스템에 쓰는 비용, 로컬 파일 시스템에서 다시 읽는 비용, 그리고 중간 결과 집합을 자식 프로세서를 운영하는 노드로 전송하는 비용을 포함해요. I/O 비용 외에도 셔플링은 CPU 비용으로 계산되어야 하는 데이터 정렬도 요구해요.

  • HDFS Read/Write는 비싸다

HDFS에 데이터를 읽고 쓰는 것은 로컬 FS보다 더 비싸요. Map-Reduce 프레임워크에서 Table Scan은 일반적으로 HDFS에서 데이터를 읽고, 두 Map-Reduce 잡 사이를 전환할 때 HDFS에 데이터를 써요. Tez에서는 모든 연산자가 단일 Tez 잡 안에서 동작해야 하므로 중간 결과 집합을 HDFS에 쓰는 비용을 지불하지 않아도 돼요.

Hive의 비용 기반 최적화는 다음 관점에서 비용을 추적해요:

  • CPU usage
  • IO Usage
  • Cardinality
  • Average size of the tuple in the relation

관계의 평균 튜플 크기와 관계의 카디널리티는 관계를 메모리에 유지하는 데 필요한 리소스를 추정하는 데 사용돼요. 관계를 유지하는 데 필요한 메모리는 Map/Bucket Join 같은 특정 조인 알고리즘을 사용할 수 있는지 결정하는 데 사용돼요.

Calcite의 Volcano 최적화기는 트리의 각 연산자 비용을 구하고 합산해 누적 비용을 찾아 두 동등 쿼리 플랜의 비용을 비교해요. 누적 비용이 가장 낮은 플랜이 쿼리를 실행할 최선의 플랜으로 선택돼요. "VolcanoCost"는 Calcite의 Volcano 최적화기의 비용을 나타내는 Java 클래스예요. "VolcanoCost" 비교 연산자는 한 비용이 다른 비용보다 작은지 결정하기 위해 행 수만 고려하는 것으로 보여요.

Hive의 경우 카디널리티를 비교하기 전에 CPU와 IO 사용을 먼저 고려하고 싶어요.

"VolcanoCost"에서 파생된 새 "RelOptCost" 구현 "HiveVolcanoCost"를 도입할 것을 제안해요. "HiveVolcanoCost"는 CPU, I/O, Cardinality, 평균 튜플 크기를 유지할 거예요. 비용 비교 알고리즘은 카디널리티에 주의를 기울이기 전에 CPU와 IO 비용에 중요성을 부여할 거예요. CPU와 IO 비용은 나노초 단위로 저장돼요. 다음은 "RelOptCost.isLe" 함수의 의사 코드예요.

Class HiveVolcanoCost extends VolcanoCost {

Double m_sizeOfTuple;

               @Override
               ...

더 알아보기 (Learn more)