맵리듀스가 아닌 일괄 처리

맵리듀스를 넘어

• 2000년대 후반 맵리듀스는 가장 인기있는 분산 시스템의 프로그래밍 모델 중 하나였지만, 데이터 양, 자료 구조, 데이터를 처리하는 방식에 따라 다른 도구가 연산을 표현하는 데 더 적합할 수 있음

• 그럼에도 불구하고 맵리듀스는 학습하기 매우 유용하며 단순함. 여기서 단순함은 무엇을 하고 있는지 이해하기 쉽다는 의미이지, 사용하기 쉽다는 의미는 아님. 맵리듀스 원시 API를 사용해서 복잡한 연산을 구현하는 일은 실제로 매우 어렵고 수고가 많이 드는 작업

• 맵리듀스를 직접 사용하는 일이 어렵기 때문에 맵리듀스 상에서 추상화된 다양한 고수준 프로그래맹 모델(피그, 하이브, 캐스캐이딩) 등이 등장

• 맵리듀스의 가장 큰 단점은 매우 견고하다는 것. 태스크가 가주 종료되어 신뢰할 수 없는 멀티 테넌트 시스템에서도 대규모 데이터를 처리하는 맵리듀스는 비록 느릴지언정 성공을 보장

• 이번 장에서는 맵리듀스 일괄 처리 방법의 대안을 살펴봄

중간 상태 구체화

• 맵리듀스 작업은 다른 작업과 모두 독립적으로 작동하며, 그 접점은 HDFS 상의 입력과 출력 디렉터리임

• 대개 한 작업의 출력이 같은 팀 내에서 유지보수하는 다른 특정 작업의 입력으로 사용된다면, 위 설정은 합리적이라고 말할 수 있음

• 이 경우 HDFS 상에 있는 파일들은 단순히 한 작업에서 다른 작업으로 데이터를 옮기는 수단, 즉 중간상태(Intermediate state)이며, 중간 상태를 파일로 기록하는 과정을 구체화(materialization)(요청이 왔을 때 계산을 시작하는 것이 아니라 미리 특정 연산 결과를 만들어 둔다는 의미)이라고 함

• 반대로 유닉스 파이프에서는 중간 상태를 완전히 구체화하는 대신 작은 인메모리 버퍼만을 사용해 점진적으로 출력을 입력으로 스트리밍함

• 중간 상태를 완전히 구체화하는 맵리듀스 접근법은 유닉스 파이프에 비해 단점이 존재

  • 맵리듀스 작업은 입력을 생성하는 모든 선행 작업이 완료됐을 때만 시작 가능함. 반면 유닉스 파이프는 동시에 시작되고 출력은 생산되는 즉시 소비됨. 이 영우 한쪽으로 쏠리는 등의 이유로 한 태스크가 병목으로 작용할 수 있음

  • 매퍼는 종종 중복될 수 있음. 즉, (맵 - 리듀스 - 맵 - 리듀스) 단계를 (맵 - 리듀스 - 리듀스) 단계로 연결할 수 있는 경우도 존재할 수 있음

  • HDFS에서 중간 상태를 저장한다는 것은 중간 상태 파일들이 여러 파일에 걸쳐 복제됐다는 의미이며, 이 것은 임시 데이터에게는 과잉조치일 수 있음

데이터플로 엔진

• 중간 상태 구체화라는 맵리듀스의 문제를 해결하기 위해 분산 일괄 처리 연산을 수행하는 엔진 몇 가지가 개발됨

• 스파크(Spark), 테즈(Tez), 플링크(Flink)가 널리 알려진 엔진임

• 구체적인 설계 방식에는 많은 차이가 있지만 공통점은 전체 워크플로를 독립된 하위 작업으로 나누지 않고 작업 하나로서 다룬다는 점임

• 이처럼 여러 처리 단계를 통해 데이터 흐름을 명시적으로 모델링하기 때문에 이러한 시스템을 데이터플로 엔진(dataflow engine)이라고 함

• 데이터플로 엔진은 입력을 파티셔닝해 병렬화 하며, 한 함수의 출력을 다른 함수의 입력으로 사용하기 위해 네트워크를 통해 복사함

• 맵리듀와는 달리 이 함수들은 맵과 리듀스를 번갈아 수행하지 않아도 되며, 유연한 방식으로 함수를 조합해서 사용할 수 있음. 이런 함수를 연산자(operator)라고 부르고, 데이터플로 엔진은 연산자의 출력과 입력을 연결하는 여러 가지 선택지를 제공

  • 레코드를 키로 재파티셔닝하고 정렬하는 것으로 맵리듀스의 셔플 단계와 비슷한 방식

  • 여러 입력을 가져와 파티셔닝하는 것은 동일하지만 정렬을 건너뛰는 선택지. 파티션 해시 조인에서 레코드를 파티셔닝하는 일은 중요하지만 해시 테이블을 사용하면 어차피 순서가 무작위로 부여되기 때문에 이 과정을 생략하는 것

  • 브로드캐스트 해시 조인을 사용한다면 한 연산자의 출력을 조인 연산자의 모든 파티션으로 보내는 방식

• 이러한 방식들은 맵리듀스 모델과 비교했을 때 다양한 장점을 가짐

  • 정렬과 같은 값비싼 작업을 실제로 필요할 때만 수행할 수 있음. 맵리듀스는 기본적으로 정렬 작업을 항상 수행함

  • 필요없는 맵 태스크를 생략할 수 있음

  • 워크플로에 모든 조인과 데이터 의존 관계를 명시적으로 선언해 스케줄러가 어느 데이터가 어디에 필요한지에 대한 개요를 가져서 지역성 최적화가 가능해짐(같은 곳에 데이터가 있다면 네트워크를 통하지 않고 공유 메모리 버퍼를 통해 교환할 수 있음)

  • 연산자 간 중간 상태를 HDFS가 아닌 메모리나 로컬 디스크에 기록해 I/O 비용을 줄일 수 있음

  • 연산자들은 입력이 준비되는 즉시 실행할 수 있음

  • 새로운 연산자를 실행할 때 이미 존재하는 JVM을 재활용할 수 있음. 맵리듀스는 각 태스크마다 새로운 JVM을 구동함

내결함성

• HDFS에 중산 상태를 구체화할 때 생기는 이점은 내구성임. 맵리듀스는 중간 상태 구체화를 통해 쉽게 내결함성을 확보함

• 하지만 스파크와 플링크, 테즈는 HDFS에 중간 상태를 쓰지 않기 때문에 내결함성 확보를 위해 다른 접근법을 사용함. 장비가 죽어서 장비에 있던 중간 상태를 잃으면 아직 유효한 데이터로부터 계산을 다시 해서 복구함(가능하다면 선행 중간 단계를 사용하고 그렇지 않다면 HDFS 상에 있는 원본 데이터를 사용함)

• 이때 재계산을 위해선 프레임워크에서 주어진 데이터가 어떻게 연산되는지 추적해야 함. 어느 입력 파티션을 사용했는지와 어떤 연산자를 적용했는지도 추적해야 함. 스파크는 데이터의 조상을 추적하기 위해 RDD(resilient distributed dataset) 추상화를 사용함(분산 변경 불가능한 객체 모음, 스파크의 모든 작업은 새로운 RDD를 만들거나 존재하는 RDD를 변형하거나 결과 계산을 위해 RDD에서 연산)

• 데이터를 재연산할 때 중요한 점은 해당 연산이 결정적인지 아닌지 파악하는 것임. 즉, 동일한 입력 데이터가 주어졌을 때 연산자들이 항상 같은 출력을 생산하는지의 여부가 중요함. 만일 이미 다운스트림으로 보낸 데이터 중 일부를 잃어버렸다면, 연산자가 비결정적이라면? 이런 경우 다운 스트림 연산자도 죽이고 신규 데이터를 기준으로 다시 수행하는 방법이 일반적임

• 따라서 이러한 상황을 방지하기위해 연산자를 결정적으로 만드는 것이 좋음(ex, 확률 통계 알고리즘에서 난수 생성시 고정된 시드를 사용하기)

구체화에 대한 논의

• 맵리듀스는 각 명령의 출력을 임시 파일에 기록하는 것과 유사한 반면 데이터플로 엔진은 유닉스 파이프와 비슷함

• 다만, 한 가지 예외가 존해함. 정렬 연산자는 출력을 생산하기 전에 전체 입력을 소비해야함. 맨 마지막 입력 레코드가 가장 낮은 키를 가져서 첫 번째 출력 레코드가 될 수도 있기 때문. 따라서 정렬이 필요한 연산자는 적어도 일시적이라도 상태를 누적해야 함

• 작업을 완료하면 출력을 다른 사용자가 찾아 사용할 수 있도록 HDFS에 기록. 다시말해 데이터플로 엔진의 구체화된 데이터셋은 보통 작업의 입력과 최종 출력 뿐임

• 정리하자면 입력은 불변이며, 최종 출력을 완전히 교체하는 방식은 맵리듀스와 비슷하지만, 모든 중간 상태를 HDFS에 기록하지 않는 차이점이 존재

그래프와 반복 처리

• 추천 엔진 같은 머신러닝이나 랭킹 시스템 분야에서 그래프 처리의 필요성이 떠오르고 있음

• 가장 유명한 그래프 분석 알고리즘 중 하나인 페이지랭크는 웹 페이지를 링크하는 다른 웹 페이지를 기반으로 인기도를 측정하는 알고리즘임

• 많은 그래프 알고리즘은 한 번에 하나의 간선을 순회하는 방식으로 표현되는데, 정점 하나와 인접한 정점을 조인하면서 특정 조건에 도달할 때까지 반복하게 됨. 이러한 알고리즘을 이행적 폐쇄(transitive closure)라고 함

• 그래프는 분산 파일 시스템에 정점과 간선 목록이 포함된 파일 형태로 저장할 수 있지만, "완료할 때까지 반복"이라는 개념은 일반적인 맵리듀스로 표현할 수 없음. 맵리듀스는 데이터를 일회성으로만 처리하기 때문이며, 이런 알고리즘은 대개 반복적 스타릴로 구현

  1. 외부 스케줄러가 이 알고리즘의 한 단계를 연산하기 위해 일괄 처리를 수행

  2. 1이 완료되면 스케줄러는 종료 조건을 기반으로 완료됐는지 확인

  3. 끝나지 않았다면 1단계로 돌아가서 일괄 처리를 수행

• 이러한 접근법을 맵리듀스로 구현할 수는 있지만 상당히 비효율적임

프리글 처리 모델

• 일괄 처리 그래프를 최적화하는 방법으로 벌크 동기식 병렬(bulk synchronous parallel, BSP) 연산 모델이 널리 사용되며 프리글(Pregel) 모델로 불림

• 구글의 프리글 논문에서 처음 시작됨

• 맵리듀스에서 매퍼가 특정 리듀서를 호출해 메시지를 전달하는 것처럼 프리글에서 한 정점은 다른 정점으로 메시지를 보낼 수 있으며 그래프의 간선을 따라 보내짐

• 반복할 때마다 개별 정점에서 함수를 호출해 그 정점으로 보내진 모든 메시지를 전달(리듀서를 호출하는 것과 비슷)

• 맵리듀스와 프리글 모델의 차이점은 정점은 반복에서 사용한 메모리의 상태를 기억하고 있다는 점이며, 따라서 정점은 새로 들어오는 메시지만 처리함

• 프리글 모델에서 정점의 상태를 제외하고 정점 사이의 메시지는 내결함성과 지속성이 있으며 메시지 통신은 고정된 횟수 안에 처리함

내결함성

• 정점이 서로 직접 질의하는 것이 아닌, 메시지 전달로 통신한다는 점은 프리글의 작업 성능 향상에 기여

• 메시지는 일괄 처리가 가능해 통신 중 대기 시간이 발생하지 않기 때문(대기 시간은 반복과 다음 반복 사이에서만 발생)

• 프리글 모델에서 앞선 반복에서 보낸 모든 메시지는 다음 반복에 도착됨을 보장하기 때문에 다음 반복을 시작하기 전에 앞선 반복은 반드시 끝나야 하며 모든 메시지는 네트우크 샇에서 복사되어야 함

• 네트워크 상의 문제로 메시지가 사라지거나 중복되거나 지연되더라도 프리글 구현상 다음 반복에서 메시지는 목적지 정점에서 정확히 한 번만 처리되며 이는 맵리듀스와 마찬가지로 프리글 프레임워크 차원에서 완벽히 결함을 복구함

• 이러한 내결함성은 반복이 끝나는 시점에 모든 정점의 상태를 주기적으로 체크포인트로 저장함으로써 보장됨. 즉, 노드에 장애가 나서 인메모리 상태가 손실되면 전체 그래프 연산을 마지막 체크포인트로 되돌려서 연산을 재시작할 수 있음(알고리즘이 결정적이고 메시지가 로그로 남는다면 손실된 파티션만 선택해서 복구할 수도 있음)

병렬 실행

• 정점은 어떤 물리 장비에서 실행되는지 알 필요가 없기 때문에 메시지를 다른 정점으로 보낼 때 단순히 정점 ID를 사용해서 메시지를 전달함

• 그래프를 파티셔닝하는 것은 프리글 프레임워크가 담당

• 이상적으로는 빈번하게 통신하는 정점끼리 같은 장비에 위치하도록 파티셔닝하는 게 좋겠지만, 그렇게 최적화하는 것은 어렵기 때문에 실제로는 단순히 임의 정점 ID를 기준으로 단순하게 분할함

• 결과적으로 그래프 알고리즘은 장비간 통신 오버헤드가 많이 발생하며, 때에 따라서 원본 그래프보다 노드 간 전송 메시지의 크기가 더 클 수도 있음

• 이런 이유로 그래프가 단일 컴퓨터 메모리에 넣을 수 있을 정도로 충분히 작다면, 단일 장비 알고리즘(ex. 그래프치(GraphChi))와 같은 단일 장비 그래프 처리 프레임워크를 쓰는 것도 좋은 선택이 될 수 있음

고수준 API와 언어

• 스파크와 플링크 같은 데이터플로 API는 일반적으로 관계형 스타일의 빌딩 블록을 사용해 연산을 표현함

• 예를 들어 특정 필드의 값을 기준으로 데이터셋을 조인하고 키로 튜플을 구룹화하고 특정 조건으로 필터링하고 튜플을 세거나 더하거나 혹은 다른 연산을 해서 집계하는 식임

• 내부적으로 이러한 연산들은 이번 장에서 설명한 다양한 조인과 그룹화 알고리즘을 사용해 구현됨

• 이러한 고수준 API의 장점은 코드를 적게 작성, 대화식 사용 지원, 생산성 향상 등이 있음

선언형 질의 언어로 전환

• 관계형 연산자로 조인을 나타내면 프레임워크가 조인 입력의 속성을 분석해 자동으로 적절한 조인 알고리즘을 선택할 수 있다는 장점이 있음

• 하이브, 스파크, 플링크는 이럭식의 질의 최적화기를 내장하고 있음

• 임의 코드를 실행할 수 있다는 것은 MPP 데이터베이스와 맵리듀스를 구별하는 특징임. 데이터베이스가 사용자 정의 함수를 지원한다고 해도 사용하기 부담스럽고, 대부분의 프로그래밍 언어의 패키지 관리자 및 의존성 관리 시스템과의 통합이 어려움

• 데이터플로 엔진도 조인 외에 좀 더 선언적인 기능을 통합하면 이점을 가질 수 있음. 예를 들어 콜백 함수에 간단한 필터링 조건만 포함되거나 레코드의 특정 필드만 선택하는 경우에 모든 레코드를 대상으로 함수를 호출하는 작업은 CPU 오버헤드가 발생함. 이를 선언적인 방법으로 표현하면 질의 최적화기가 칼럼 기반 저장 레이아웃을 이용해 필요한 칼럼만 읽을 수 있음

• 하이브와 스파크의 데이터 프레임(DataFrame)은 딱 맞는 내부 루프에서 데이터를 반복해서 CPU 캐시가 잘되게 하거나 함수 호출을 피하는 방식을 사용함

• 고수준 API에 선언적 측면을 포함하면서 질의 최적화기를 가진다면 일괄 처리 프레임워크는 MPP 데이터와 한층 더 비슷해지며, 임의 코드를 실행하고 임의 형식의 데이터를 읽을 수 있다는 확장성을 지니며 일괄 처리 프레임워크의 장점인 유연성은 그대로 유지할 수 있음

다양한 분야를 지원하기 위한 전문화

• 임의 코드를 실행할 수 있는 확장성은 매우 우용하지만 표준화된 처리 패턴이 계속 나타나는 경우도 많음

• 따라서 재사용 가능한 공통 블록을 구현(주로 통계학이나 수치 알고리즘). 이를 테면 머하웃(Mahout)은 맵리듀스, 스파크, 플링크 상에서 실행되는 다양한 머신러닝용 알고리즘 구현을 가지고 있음

• 일괄 처리 엔진은 점차 광범위한 영역에서 필요한 알고리즘을 분산 수행하는 데 사용됨. 일괄 처리 시스템은 내장 기능과 고수준 선언적 연산자를 모두 가지고 있고 MPP 데이터베이스는 한층 프로그래밍이 가능하게끔 유연해졌기 때문에 이 둘은 더욱더 비슷해지고 있음