Hive on Spark

Hive on Spark

Hive에 Spark를 세 번째 실행 백엔드(execution backend)로 추가하는 설계 문서예요. MapReduce, Tez와 병렬로 Spark에서 Hive 쿼리를 실행할 수 있게 하는 것이 목표입니다.

출처: 문서

본문

1. 소개 (Introduction)

Hive를 수정해 Spark를 MapReduce와 Tez에 병렬하는 세 번째 실행 백엔드로 추가하는 것을 제안합니다 (HIVE-7292). Spark는 Hadoop의 2단계 MapReduce 패러다임 밖이지만 HDFS 위에 구축된 오픈소스 데이터 분석 클러스터 컴퓨팅 프레임워크예요.

Spark의 주요 추상화는 Resilient Distributed Dataset(RDD)이라 부르는 아이템들의 분산 컬렉션입니다. RDD는 Hadoop InputFormat(예: HDFS 파일)에서 만들거나, 다른 RDD를 변환해서 만들 수 있어요. groupBy·filter 같은 일련의 변환과 count·save 같은 액션을 적용해 RDD를 처리·분석하면 중간 단계 없이 MapReduce 작업이 하는 일을 수행할 수 있습니다. SQL 쿼리는 Shark와 Spark SQL에서 보여주듯 쉽게 Spark 변환과 액션으로 옮길 수 있어요. 실제로 join·count 같은 많은 기본 변환·액션들이 SQL 지향적입니다.

Spark에 대한 추가 정보:

1.1 동기 (Motivation)

Hive를 Spark에서 실행하게 하는 주요 동기는 다음과 같아요.

  • Spark 사용자에게 이점: 이미 다른 데이터 처리·머신러닝 요구로 Spark를 사용하는 사용자에게 매우 유용합니다. 하나의 실행 백엔드로 통일하면 운영 관리가 편리하고, 문제 디버깅·개선을 위한 전문성을 키우기도 쉬워요.
  • Hive 채택 확대: 이것은 Hive를 Spark 사용자 기반에 SQL on Hadoop 옵션으로 끌어들이며 Hive의 채택을 더 늘립니다.
  • 성능: 여러 reducer 단계를 포함하는 Hive 쿼리가 Tez처럼 더 빨리 실행되며 사용자 경험을 개선합니다.

Spark 실행 백엔드가 Tez나 MapReduce를 대체하는 것은 목표가 아니에요. 여러 백엔드가 공존하는 것이 Hive 프로젝트에 건강합니다. 사용자는 Tez, Spark, MapReduce 중 무엇을 쓸지 선택할 수 있고, 각각 사용 사례에 따라 다른 강점이 있습니다. Hive의 성공은 Tez나 Spark 어느 한쪽의 성공에 완전히 의존하지 않아요.

1.2 설계 원칙 (Design Principle)

주된 설계 원칙은 Hive의 기존 코드 경로에 영향이 없거나 제한적이어서 기능적·성능적 영향이 없게 하는 것입니다. 즉 MapReduce나 Tez에서 Hive를 실행하는 사용자는 오늘날과 같은 기존 기능과 코드 경로를 갖게 됩니다.

또한 실행 계층에 Spark를 끼워 넣으면 코드 공유를 최대화하고 유지보수 비용을 묶어두므로, Hive 커뮤니티가 Spark를 위해 특수한 투자를 할 필요가 없어요. 한편 Spark를 실행 엔진으로 선택한 사용자는 자동으로 Hive가 제공하는 모든 풍부한 기능을 얻습니다. Hive에 추가된 미래 기능(새 데이터 타입, UDF, 논리 최적화 등)이 Hive의 Spark 실행 엔진에서 커스터마이징 없이 자동으로 사용 가능해야 해요.

1.3 Shark 및 Spark SQL과의 비교 (Comparison with Shark and Spark SQL)

Spark 생태계에는 Spark에서 Hive QL을 지원하는 두 개의 관련 프로젝트가 있습니다. 바로 Shark와 Spark SQL이에요.

  • Shark 프로젝트는 Hive가 만든 쿼리 플랜을 자체 표현으로 변환해 Spark 위에서 실행합니다.
  • Spark SQL은 Spark의 기능입니다. Hive QL 지원을 위해 Hive의 파서를 프론트엔드로 사용해요. Spark 애플리케이션 개발자는 코드에서 다른 Spark 연산자뿐 아니라 SQL로도 쉽게 데이터 처리 로직을 표현할 수 있습니다. Spark SQL은 Hive와 다른 사용 사례를 지원해요.

Shark와 Spark SQL에 비해 우리의 접근 방식은 설계상 모든 기존 Hive 기능(Hive QL과 미래 확장 포함)과 Hive의 인증·모니터링·감사·기타 운영 도구 통합을 지원합니다.

1.4 기타 고려사항 (Other Considerations)

새 실행 백엔드는 큰 일입니다. 설계가 기존 코드 경로를 건드리지 않더라도 필연적으로 복잡성과 유지보수 비용을 더해요. 이제 Hive는 MapReduce, Tez, Spark에 대해 유닛 테스트를 갖게 됩니다. 이점이 비용보다 크다고 생각합니다.

인프라 관점에서 지속적 통합을 위한 하드웨어 후원을 받을 수 있습니다. 마지막으로 Hive on Tez는 Spark 같은 새 실행 엔진을 지원하는 데 매우 유용한 중요한 기반 작업을 마련했어요.

한편 Spark는 MapReduce나 Tez와 매우 다른 프레임워크입니다. 따라서 통합 중에 갭과 문제가 나타날 가능성이 높아요. Hive 커뮤니티가 Spark 커뮤니티와 긴밀히 협력해 통합을 성공시키리라 기대합니다.

2. 상위 수준 기능 (High-Level Functionality)

2.1 새 실행 엔진 (A New Execution Engine)

기존의 MapReduce와 Tez에 더해 실행 엔진 Spark를 새로 도입합니다. Hive에서 Spark를 실행 엔진으로 사용하려면 다음을 설정하세요.

set hive.execution.engine=spark;

이 구성의 기본값은 여전히 "mr"입니다. Spark가 없는 클러스터에서는 Hive가 여전히 MapReduce와 Tez 위에서 그대로 동작해요. 새 실행 엔진은 쿼리를 전혀 수정하지 않고 모든 Hive 쿼리를 지원해야 하며, 쿼리 결과는 MapReduce나 Tez의 결과와 기능적으로 동일해야 합니다.

2.2 Spark 구성 (Spark Configuration)

Spark가 Hive의 실행으로 구성되면 Spark 클러스터의 master URL 같은 몇 가지 구성 변수가 도입됩니다. 하지만 Spark가 실행 엔진으로 구성되지 않으면 완전히 무시될 수 있어요.

2.3 기타 기능 (Miscellaneous Functionality)

Hive는 MapReduce와 Tez의 explain 명령에 표시되는 것과 유사한 태스크 실행 플랜을 표시합니다. Spark에서 쿼리를 실행할 때 Hive는 쿼리의 진행 및 완료 상태에 대한 적절한 피드백을 사용자에게 줍니다. 사용자는 이전과 같이 통계·진단 정보(콘솔의 counters, 로그, 디버그 정보)를 얻을 수 있어요.

3. Hive 레벨 설계 (Hive-Level Design)

소개에서 언급했듯 이 프로젝트는 Shark나 Spark SQL과 다른 접근 방식을 취해요. Spark의 프리미티브로 SQL 시맨틱을 구현하지 않는다는 뜻입니다. 반대로 MapReduce 프리미티브로 구현할 거예요. 여기서 새로운 것은 이 MapReduce 프리미티브를 Spark에서 실행한다는 점뿐입니다. 실제로 이 설계에서는 Spark 프리미티브 중 극히 일부만 사용됩니다.

Shark나 Spark SQL과 다른 방식으로 Hive의 MapReduce 프리미티브를 Spark에서 실행하는 접근은 다음과 같은 직접적인 장점이 있어요.

  • Spark 사용자는 Hive가 미래에 도입할 새 기능을 포함한 전체 풍부한 기능 세트를 자동으로 얻습니다.
  • 이 접근은 Hive의 Spark 실행 엔진에서 커스터마이징 작업의 필요성을 피하거나 줄여줍니다.
  • Hive-on-Spark를 Hive MapReduce·Tez와 동일하게 유지함으로써 프로젝트 범위를 제한하고 장기 유지보수를 줄여줍니다.

Hive용 Spark 실행 엔진을 구현하는 주요 작업은 두 부분입니다. (1) 쿼리 플래닝(query planning) – 시맨틱 분석기의 Hive operator 플랜을 더 나아가 Spark가 실행할 수 있는 태스크 플랜으로 변환, (2) 쿼리 실행(query execution) – 생성된 Spark 플랜을 실제로 Spark 클러스터에서 실행. 물론 모니터링, 카운터, 통계 같은 기타 잡다하지만 없어서는 안 될 기능 파트들도 있어요.

Spark가 주로 Scala로 작성됐지만 Java를 포함한 여러 언어로 클라이언트 API를 제공한다는 점이 중요합니다. 자연스럽게 통합에는 Spark Java API를 선택하며, 이 프로젝트에 Scala 지식은 필요 없어요.

3.1 쿼리 플래닝 (Query Planning)

현재 주어진 사용자 쿼리에 대해 Hive 시맨틱 분석기는 TableScanOperator, ReduceSink, FileSink, GroupByOperator 같은 논리 연산자 그래프로 구성된 operator 플랜을 생성합니다. MapReduceCompiler는 논리 operator 플랜에서 MapReduceTasks와 기타 헬퍼 작업(MoveTask 등)의 그래프를 컴파일해요. Tez는 비슷하지만, 여러 MapReduce 태스크를 단일 Tez 태스크로 결합하는 TezTask를 생성합니다.

Spark의 경우 MapReduceCompiler와 TezCompiler에 평행한 SparkCompiler를 도입합니다. 그 주요 책임은 Hive 논리 operator 플랜에서 Spark에서 실행 가능한 플랜을 컴파일하는 것입니다. 따라서 Spark 클러스터에서 실행될 작업을 나타내는 SparkTask와 Spark 태스크의 플랜을 설명하는 SparkWork를 갖게 됩니다. SparkCompiler는 Hive의 operator 플랜을 SparkWork 인스턴스로 변환합니다.

태스크 플랜 생성 동안 SparkCompiler는 Spark에 적합한 물리적 최적화를 수행할 수 있어요. 다만 구현의 첫 단계에서는 쉽고 명확한 경우가 아니면 여기에 집중하지 않겠습니다. Spark에 대한 지식·경험이 쌓이면서 증분적으로 추가 최적화를 할 수 있습니다.

Hive의 operator 플랜에서 SparkWork를 어떻게 생성할지는 구현에 맡겨집니다. 하지만 Tez와 Spark, MapReduce와 Spark 사이에 공통 로직이 많아 보여요. 가능하면 공통 로직을 추출해 공유 가능한 형태로 패키징하고, 각 태스크 컴파일러에 특정 구현을 맡겨 MapReduce나 Tez를 불안정하게 만들지 않도록 합니다.

3.2 작업 실행 (Job Execution)

SparkTask 인스턴스는 다른 태스크와 같은 방식으로 Hive의 태스크 실행 프레임워크가 실행할 수 있어요. 내부적으로 SparkTask.execute() 메서드는 SparkWork 인스턴스로 RDD와 함수를 만들고, Spark 클라이언트를 통해 실행을 Spark 클러스터에 제출합니다.

Spark 작업이 Spark 클러스터에 제출되면 Spark 클라이언트는 작업 실행을 계속 모니터링하고 진행을 보고합니다. Spark 작업은 SparkListener API를 통해 모니터링할 수 있어요. 현재 Spark Java API에는 없지만, Spark 커뮤니티의 도움으로 곧 제공되리라 기대합니다. SparkListener API로, MapReduce 처리에 쓰는 HadoopJobExecHelper나 Tez 작업 처리에 쓰는 TezJobMonitor와 비슷한 기능을 제공하는 SparkJobMonitor 클래스를 추가할 거예요. 이 클래스는 작업 실패 시 실행 시점에 던져진 최상위 예외도 검색해 출력합니다.

Spark 작업 제출은 사용자 구성으로 인스턴스화된 SparkContext 객체를 통해 이뤄져요. Hive가 SparkTask를 실행하면 그런 컨텍스트 객체가 현재 사용자 세션에 생성됩니다. 컨텍스트 객체로 Hive 테이블에 해당하는 RDD가 만들어지고, Hive의 SparkWork로 빌드된 MapFunction과 ReduceFunction이 RDD에 적용됩니다. 작업 실행은 RDD에 더미 함수로 foreach() 변환을 적용해 트리거됩니다.

사용자 세션당 하나의 SparkContext가 맞지만, Spark는 스레드 안전성 문제 때문에 애플리케이션당 하나의 SparkContext를 가정하는 것 같습니다. Spark 커뮤니티가 이 문제를 시의적절하게 해결하리라 기대해요.

3.3 설계 고려사항 (Design Considerations)

이 섹션은 새로 도입되거나 특별한 처리가 필요한 기존 컴포넌트 등 여러 중요한 컴포넌트의 주요 설계 고려사항을 다룹니다. UDF나 커스텀 Serde처럼 이름이 거론되지 않은 다른 기존 컴포넌트는 특별한 고려가 필요 없거나 무시할 수준이라고 기대합니다.

Table as RDD

Hive 테이블은 HDFS의 파일·폴더 묶음일 뿐이에요. Spark 프리미티브는 RDD에 적용됩니다. 따라서 자연스럽게 Spark 실행 엔진에서 Hive 테이블을 RDD로 취급해요. 하지만 Hive 테이블은 HDFS 파일보다 복잡합니다. 파티션과 버킷을 가질 수 있고, 이종 입력 형식과 스키마 진화를 다룹니다. 결과적으로 처리가 그리 간단하지 않을 수 있고, 잠재적 복잡성이 있어 이를 인지해야 해요. Spark의 Hadoop RDD를 확장하고 Hive 특화 RDD를 구현해야 할 수도 있습니다. RDD 확장은 Scala에서 쉬워 보이지만 Spark의 Java API에 그런 능력이 없어 까다로울 수 있어요. RDD 확장이 필요한지 알아보고, 필요하다면 Java API에 대해 Spark 커뮤니티의 도움을 받아야 합니다.

SparkWork

앞서 논의했듯 SparkTask는 Spark 작업이 실행할 태스크 플랜을 설명하는 SparkWork를 사용합니다. SparkWork는 TezWork와 매우 유사할 텐데, 기본적으로 리프에는 MapWork, 다른 모든 노드에는 ReduceWork(가끔 UnionWork)로 구성됩니다. SparkWork를 MapWork·ReduceWork로 정의하면 새 개념을 더 쉽게 이해할 수 있어요. explain 명령은 Hive 사용자에게 익숙한 패턴을 보여줄 것입니다.

SparkTask

SparkWork 인스턴스가 설명하는 작업을 실행하려면 추가 변환이 필요해요. MapWork·ReduceWork는 MapReduce 지향 개념이라, Spark로 구현하려면 플랜을 탐색하고 Spark 구조물(RDD, 함수)을 생성해야 합니다. 탐색·변환 방법은 구현에 맡기되, 이것은 매우 Spark 특화적이라 다른 컴포넌트에 노출되거나 영향을 주지 않아요.

앞서 언급한 MapFunction은 MapWork, 특히 ExecMapper.map() 메서드에서 시작하는 operator 체인으로 만들어집니다. ExecMapper 클래스는 MapReduce Mapper 인터페이스를 구현하지만, Hive의 구현에는 Spark에서 재사용할 수 있는 코드가 있습니다. 따라서 공통 코드를 별도 클래스 MapperDriver로 추출해 MapReduce와 Spark가 공유할 가능성이 높아요. 이것은 재설계가 아니라 리팩터링일 뿐입니다. (Tez도 비슷한 상황이었지만, Tez는 RecordProcessor라는 별도 클래스를 만들어 비슷한 역할을 했습니다.)

비슷하게 ReduceFunction은 SparkWork의 ReduceWork 인스턴스로 만들어질 거예요. Spark에게 ReduceFunction은 MapFunction과 다를 바 없지만, 함수 구현은 ExecReducer.reduce()에서 시작하는 operator 체인으로 달라집니다. ExecReducer의 일부 코드가 재사용되므로, 공통 코드를 ReducerDriver라는 별도 클래스로 추출해 MapReduce와 Spark가 공유할 가능성이 높습니다.

MapFunction과 ReduceFunction을 포함한 모든 함수는 직렬화 가능해야 해요. Spark가 그것들을 클러스터로 보내야 하기 때문입니다. 이것은 함수 패키징 방식이 함수의 직렬화에 영향을 주고 Spark가 이에 대해 암묵적이어서 까다로울 수 있습니다.

Spark의 내장 메서드 map, reduce 변환 연산자는 각 레코드에 대해 함수적입니다. 하지만 Hive 연산자는 행을 처리하도록 호출되기 전에 초기화되어야 하고 처리가 끝나면 닫혀야 해요. MapFunction·ReduceFunction은 이 모든 것을 단일 call() 메서드에서 수행해야 합니다.

Hive의 대체 실행 백엔드로 Spark를 사용하기 위해 RDD에 mapPartitions 변환 연산자를 사용할 거예요. 이것은 데이터 전체 파티션에 대한 반복자(iterator)를 제공합니다. 반복자를 제어하면 Hive는 첫 행을 처리하기 전에 operator 체인을 초기화하고 모든 입력이 소비된 후 초기화를 해제할 수 있어요. 프로토타이핑 중 Spark가 어떤 경우 함수를 전역적으로 캐시해 함수의 오래된 상태를 유지한다는 점이 주목할 만합니다. 이런 원인은 잡기 어렵고, Spark가 향후 기능을 더 명확히 문서화하길 바랍니다.

셔플, 그룹, 정렬 (Shuffle, Group, and Sort)

이것은 MapReduce와 Tez에서 "공짜"로 오지만, Spark에 동등한 것을 제공해야 해요. 다행히 Spark는 partitionBy, groupByKey, sortByKey처럼 MapReduce의 셔플 능력을 대체하기에 적합한 몇 가지 변환을 제공합니다.

  • partitionBy 변환은 순수 셔플(그룹핑·정렬 없음)을 합니다.
  • groupByKey는 셔플과 그룹핑을 합니다.
  • sortByKey()는 셔플과 정렬을 합니다.

따라서 SparkWork의 각 ReduceSinkOperator에 대해 이 변환 중 하나를 주입해야 해요. 정확한 셔플 동작을 선택적으로 고를 수 있는 능력은 최적화 기회를 줍니다. 예를 들어 Hive의 groupBy는 키가 정렬될 것을 요구하지 않지만, MapReduce는 그래도 정렬합니다. Spark에서는 키 순서가 중요할 때만(예: SQL order by) sortByKey를 선택할 수 있어요. sortByKey는 그룹핑을 제공하지 않지만, 같은 키를 가진 행들이 연속으로 오므로 키를 쉽게 그룹핑할 수 있습니다. 반면 groupByKey는 키를 컬렉션에 모으는데, 이것은 자연스럽게 MapReduce의 reducer 인터페이스에 맞습니다.

Hive가 join처럼 직접 못 쓰는 연산을 구현하기 위해 MapReduce 키를 더 정교하게 사용하므로, 위 변환들이 Hive가 필요로 하는 대로 정확히 동작하지 않을 수 있어요. 진행하면서 잠재적 문제를 식별하는 데 주의를 기울여야 합니다. 마지막으로 Spark 커뮤니티가 셔플 관련 API를 개선/변경하는 중인 것 같아서, 이 설계 부분은 변경될 수 있습니다. 자세한 내용은 https://issues.apache.org/jira/browse/SPARK-2044 를 참고하세요.

Join

MapReduce 세계에서 join을 구현하는 것은 Hive에서 드러나듯 상당히 복잡해요. Hive는 reduce-side join과 map-side join(map-side hash lookup, map-side sorted merge 포함)을 갖고 있습니다. Hive의 join 구현을 유지할 거예요. 다만 Hive가 reduce-side join 구현에서 MapReduce의 셔플을 광범위하게 사용하므로, 셔플 동작(키 생성, 파티셔닝, 정렬 등)에 특히 주의해야 합니다. 앞서 셔플·그룹·정렬 섹션에서 지적했듯 Spark가 셔플에 대한 유연한 제어를 제공하거나 제공할 것이라 기대합니다. 자세한 설계는 Hive on Spark: Join Design 문서를 참고하세요.

작업 수 (Number of Tasks)

앞서 지정했듯 partitionBy 같은 Spark 변환을 사용해 mapper 쪽 연산과 reducer 쪽 연산을 연결합니다. 그 변환들에 파티션 수를 선택적으로 줄 수 있는데, 이는 기본적으로 reducer 수를 결정합니다. reducer 수의 결정은 MapReduce와 Tez와 동일할 거예요.

로컬 MapReduce 작업 (Local MapReduce Tasks)

로컬 작업을 Spark에서 실행하는 이점(데이터를 파일로 내리고 다시 파일에서 메모리로 읽는 것을 피하는 것)을 볼 수 있지만, 단기적으로는 그런 작업을 오늘날과 같은 방식으로 실행합니다. 즉 Hive는 로컬로 실행할 때 항상 MapReduce 작업을 제출해야 한다는 뜻입니다. 그러나 이것은 추후 조사·평가할 수 있습니다.

쿼리 결과를 사용자에게 보여주는 것에도 동일하게 적용됩니다. 현재 클라이언트 쪽의 fetch 연산자가 (쿼리 플랜의 FileSink가 만든) 임시 파일에서 행을 가져옵니다. FileSink가 대신 인메모리 RDD를 생성하고 fetch 연산자가 RDD에서 행을 직접 읽게 하는 것도 가능해요. 이것 역시 향후 작업으로 조사·구현할 수 있습니다.

시맨틱 분석과 논리 최적화 (Semantic Analysis and Logical Optimizations)

시맨틱 분석기와 어떤 논리 최적화도 변하지 않습니다. 물리적 최적화와 MapReduce 플랜 생성은 이미 Hive on Tez 작업의 일부로 별도 클래스로 옮겨졌어요.

작업 진단 (Job Diagnostics)

기본 "job succeeded/failed"와 진행 상황은 "Job monitoring"에서 논의한 대로 이루어집니다. 실패한 작업에 대한 추가 정보를 가져오는 Hive의 현재 방식은 즉시 사용하지 못할 수 있지만, 이것은 더 연구가 필요한 영역입니다.

Spark는 각 SparkContext가 실행되는 동안 WebUI를 제공합니다. 이 정보는 기본적으로 애플리케이션 기간 동안에만 사용 가능해요. 사후에 웹 UI를 보려면 애플리케이션 시작 전에 spark.eventLog.enabled를 true로 설정하세요. 이것은 UI에 표시되는 정보를 인코딩한 Spark 이벤트를 영속 스토리지에 기록하도록 Spark를 구성합니다.

Spark의 Standalone 모드 클러스터 매니저도 자체 웹 UI를 갖습니다. 애플리케이션이 수명 동안 이벤트를 기록했다면, Standalone master의 웹 UI는 애플리케이션이 끝난 후 그 UI를 자동으로 다시 렌더링합니다. Spark가 Mesos나 YARN에서 실행되면, 애플리케이션의 이벤트 로그가 존재한다면 Spark의 history server를 통해 완료된 애플리케이션의 UI를 재구성할 수 있습니다. Spark 모니터링에 대한 자세한 내용은 http://spark.apache.org/docs/latest/monitoring.html 을 참고하세요.

카운터와 메트릭 (Counters and Metrics)

Spark에는 어소시에이티브 연산으로만 "더해지는" 변수인 accumulator가 있어 병렬로 효율적으로 지원될 수 있어요. 이것은 (MapReduce처럼) 카운터나 합계를 구현하는 데 쓸 수 있습니다. Spark는 수치 값 타입과 표준 변경 가능 컬렉션의 accumulator를 네이티브로 지원하며, 프로그래머가 새 타입 지원을 추가할 수 있어요. Hive에서는 Spark accumulator로 Hadoop 카운터를 구현할 수 있지만, 바로 하지는 못할 수 있습니다.

Spark는 실행 중인 작업의 런타임 메트릭을 게시합니다. 하지만 그 메트릭은 MapReduce나 Tez와 다를 가능성이 매우 높고, 메트릭 추출 방식은 말할 것도 없어요. 이 주제는 별도 문서가 필요하지만, 확실히 증분적으로 개선할 수 있습니다.

Explain 문 (Explain Statements)

Explain 문은 TezWork와 유사할 것입니다.

Hive 변수 (Hive Variables)

Hive 변수는 오늘날처럼 계속 동작합니다. 변수는 이전처럼 실행 엔진으로 전달됩니다. 다만 일부 실행 엔진 관련 변수는 Spark에 적용되지 않을 수 있으며, 그 경우 그냥 무시됩니다.

Union

UnionWork는 Spark 실행 엔진에서 MapReduce 프리미티브로 SQL 시맨틱을 구현하겠다고 위에서 언급했지만, union은 예외예요. MapReduce 프리미티브로 구현할 수는 있지만, 두 데이터셋을 union하려면 MapReduce 작업이 최대 3개 필요해요. Spark의 union 변환을 사용하면 실행 시간을 크게 줄이고 상호작용성을 높일 수 있습니다. 실제로 Tez는 union에 대해 이미 MapReduce 관행에서 벗어났어요. 기존 UnionWork가 있으며 union 연산자가 작업 단위로 변환됩니다.

동시성과 스레드 안전성 (Concurrency and Thread Safety)

Spark는 하나의 JVM에서 워커가 여러 HDFS 스플릿을 처리할 수 있다는 점에서 MapReduce와 다르게 mapper·reducer를 실행해요. 하지만 Hive의 map-side operator 트리나 reduce-side operator 트리는 단일 스레드로 배타적 JVM에서 동작합니다. operator 트리를 재사용하고 서로 공유 JVM에 두는 것은 동시성·스레드 안전성 문제를 거의 확실히 일으킵니다. ExecMapper.done 같은 변수가 초기 프로토타이핑에서 문제를 드러냈어요. 두 ExecMapper 인스턴스가 단일 JVM에 있으면, 먼저 끝나는 mapper가 다른 mapper까지 조기에 종료시킬 수 있습니다. 이 operator 트리들을 스레드 안전하고 경합이 없게 만드는 데 상당한 작업이 필요하리라 기대합니다. 하지만 이 작업은 다른 실행 엔진에 어떤 영향도 주지 않아야 합니다.

빌드 인프라 (Build Infrastructure)

Spark에 대한 새 ql 의존성이 생깁니다. 현재 Spark 클라이언트 라이브러리는 단일 jar로 제공됩니다. Spark jar는 Hadoop jar가 처리되는 것과 같은 방식으로 처리됩니다. 컴파일 중에는 사용되지만 최종 배포에는 포함되지 않아요. 대신 별도로 설치되는 것에 의존합니다. Spark jar는 Spark 작업 실행에만 있으면 되고, MapReduce나 Tez 실행에는 필요 없습니다.

한편 Hive 코드를 Spark에서 실행하려면, 특정 Hive 라이브러리와 그 의존성을 SparkContext.addJar() 메서드를 호출해 Spark 클러스터로 배포해야 해요. Spark도 Hadoop과 다른 라이브러리에 의존하는데, 이것들이 다른 버전으로 Hive의 의존성에 있을 수 있어 라이브러리 충돌을 식별·해결하는 데 어려울 수 있습니다. 프로토타이핑 동안 Jetty 라이브러리가 그런 도전을 제기했어요.

미니 Spark 클러스터 (Mini Spark Cluster)

master URL에 "local"을 주면 Spark 작업을 로컬로 실행할 수 있습니다. 대부분의 테스트가 이 모드에서 수행될 거예요. 동시에 Spark는 로컬 머신에서 주어진 수의 프로세스로 만들어진 로컬 클러스터에서 작업을 실행하는 방법도 제공해요. 이것이 Hive의 Spark 관련 테스트를 실행하는 좋은 방법인지 더 결정할 것입니다.

테스트 (Testing)

pre-commit 테스트를 포함한 테스트는 Tez와 동일합니다. 현재 Hive에는 Tez vs MapReduce, vectorization on/off 같은 전체 회귀 스위트 실행이 필요한 변수가 몇 개 있어 커버리지 문제가 있습니다. 테스트 시간을 늘리지 않으면서 충분한 커버리지를 두기 위해 pre-commit 테스트 실행에서 그 변수들을 회전시키는 것을 제안합니다.

3.4 Spark에서 잠재적으로 필요한 작업 (Potentially Required Work from Spark)

프로토타이핑·설계 동안 몇 가지 Spark 이슈가 식별됐는데, 문서 전반에 나타나 있습니다. 더 있을 수 있으나, 프로젝트를 위해 Spark 커뮤니티에서 필요한 개선의 요약은 다음과 같아요.

  • Java용 작업 모니터링 API
  • SparkContext 스레드 안전성 문제
  • 셔플 기능과 API 개선
  • 잠재적으로 RDD 확장을 위한 Java API

4. 요약 (Summary)

위 분석에서 볼 수 있듯 Spark on Hive 프로젝트는 기능·설계 측면에서 단순하고 깔끔하지만, 구현은 복잡하고 시간·자원이 크게 들 수 있어요. 따라서 단계적 접근을 취할 것이며, 모든 기본 기능은 첫 단계에 있되 최적화·개선 작업은 상대적으로 긴 기간 동안 지속되리라 기대합니다.

둘째, Hive와 Spark의 통합이 항상 매끄럽진 않을 것이라 기대해요. 기능적 갭이 식별되고 문제가 발생할 수 있습니다. Hive 커뮤니티와 Spark 커뮤니티가 가는 길의 어떤 장애물도 해결하기 위해 긴밀히 협력하리라 예상합니다.

그럼에도 기존 코드 경로에 대한 영향은 최소일 것이라 믿습니다. Spark 실행 엔진이 안정화되는 데 시간이 걸리더라도 MapReduce와 Tez는 그대로 계속 동작해야 합니다.

더 알아보기 (Learn more)

Hive on Spark는 쿼리 실행 엔진을 hive.execution.engine=spark로 바꾸기만 하면 쓸 수 있는 백엔드예요. Hive의 기존 MapReduce·Tez 백엔드와 코드 경로를 공유하도록 설계되어, Spark 사용자에게는 Hive의 모든 기능을, Hive 사용자에게는 더 빠른 실행을 제공합니다.