맵리듀스(1)

맵리듀스와 분산 파일 시스템

• 맵리듀스는 유닉스 도구와 비슷하지만, 수천 대의 장비로 분산해서 실행이 가능하다는 점에서 차이가 존재

• 유닉스 도구와 마찬가지로 입력을 수정하지 않기 때문에 출력을 생산하는 것 외에 다른 부수 효과는 없음

• 유닉스는 stdin과 stdout을 입력과 출력으로 사용하지만, 맵리듀스는 분산 파일 시스템(HDFS) 상의 파일을 입력과 출력으로 사용

• HDFS 외에도 GlusterFS, QFS 등 다양한 분산 파일 시스템이 있으며, 아마존 S3, 애저 Blob 저장소, 오픈스택 스위프트 같은 객체 저장소도 여러 면에서 유사함

  • HDFS는 필요한 파일의 복사본이 저장된 장비에서 연산을 수행(지역성)하지만, 객체 저장은 저장소와 연산 장치가 분리된다는 점에서 가장 큰 차이가 있음

• HDFS는 비공유 원칙을 기반으로 하며, 이는 NAS(Network Attached Storage)나 SAN(Storage Area Network) 아키텍처에서 사용하는 공유 디스크 방식과 반대. 그렇기 때문에 특별한 하드웨어 장비가 필요없음

• HDFS는 각 장비에서 실행된 데몬 프로세ㅔ슬 구성되며, 네임노드라는 중앙 서버에서 특정 파일 블록이 어떤 장비에 저장됐는지 추적하는 방식으로 작동

• HDFS는 이러한 특징을 통해 수만 대의 장비를 묶어 실행할 수 있으며, 범용 하드웨어와 오픈소스 소프트웨어의 조합으로 저렴한 대규모 확장을 가능케 함

맵리듀스 작업 실행하기

• 맵리듀스는 HDFS 위에서 대용량 데이터셋을 처리하는 코드를 작성하는 프로그래밍 프레임워크

• 맵리듀스의 데이터 처리 패턴은 로그 분석 예제와 유사한 부분이 많음

  1. 입력 파일을 읽고, 레코드로 쪼갬
  2. 입력 레코드마다 매퍼 함수를 호출해 키와 값을 추출(맵 단계)
  3. 키를 기준으로 키-값 쌍을 모두 정렬
  4. 정렬된 키-값 쌍 전체를 대상으로 리듀스 함수를 호출. 이 명령은 키가 같으면서 인접한 레코드의 수를 셈(리듀스 단계)

• 맵리듀스 작업 하나는 4가지 단계로 수행되며, 맵과 리듀스 단계는 사용자가 직접 작성하는 데이터 처리 코드

  • 매퍼(Mapper)

    • 매퍼는 모든 입력 레코드마다 한 번씩 호출
    • 입력 레코드로부터 키와 값을 추출하는 작업
    • 입력 레코드로부터 다음 레코드까지 상태를 유지하지 않기 때문에 각 레코드를 독립적으로 처리

  • 리듀서(Reducer)

    • 매퍼가 생산한 키-값 쌍을 받아 같은 키를 가진 레코드를 모으며, 해당 값의 집합을 반복
    • 출력 레코드를 생산

• 단순하게 표현하자면, 매퍼는 정렬에 적합한 형태로 데이터를 준비하는 역할을 수행하며, 리듀스는 정렬된 데이터를 가공하는 역할을 수행

맵리듀스의 분산 실행

• 이전에 이야기했듯 유닉스 명령어 파이프라인과 맵리듀스의 가장 큰 차이점은 맵리듀스는 여러 장비에서 병렬로 동시에 처리가 가능하다는 점이며, 이는 맵리듀스 프레임워크가 장비 간에 데이터가 이동하는 복잡한 부분을 처리하기 때문

• 분산 연산에서 매퍼와 리듀서로 표준 유닉스 도구를 사용하는 것도 가능하지만, 대개는 일반적인 프로그래밍 언어로 함수를 작성함

• 맵리듀스 작업의 병렬 실행은 파티셔닝을 기반으로 하며, 작업 입력으로 HDFS 상의 디렉터리를 사용하는 것이 일반적임

• 각 입력 파일은 보통 수백 메가바이트에 달하며, 매퍼 입력 파일의 복제본이 있는 장비에 RAM과 CPU에 여유가 충분하다면, 맵리듀스는 입력 파일이 있는 장비에서 작업을 수행하려 함(지역성)

• 작업이 시작되면 작업이 할당된 장비에 실행될 애플리케이션 코드가 복사되며, 복사가 끝나면 매퍼 태스크가 시작되고, 입력 파일을 읽어 한 번에 레코드 하나씩을 매퍼 콜백 함수(다른 코드의 인수로서 넘겨주는 실행 가능한 코드)로 전달

• 리듀서 연산도 파티셔닝됨. 맵 태스크 수는 입력 파일의 블록 수로 결정되지만, 리듀스 태스크는 사용자가 설정함. 이때 특정 키-값 쌍이 어느 리듀스 태스크에서 수행될지 결정하기 위해 키의 해시값을 사용

• 매퍼가 입력 파일을 읽어서 정렬된 출력 파일을 기록하기를 완료하면 맵리듀스 스케줄러는 그 매퍼에서 출력 파일을 가져올 수 있다고 리듀서에게 알리고, 리듀서는 각 매퍼와 연결해서 리듀서가 담당하는 파티션에 해당하는 정렬된 키-값 쌍 파일을 다운로드 함(이 과정을 셔플이라고 함)

• 리듀스 태스크는 매퍼로부터 파일을 가져와 정렬된 순서를 유지하면서 병합

맵리듀스 워크플로

• 맵리듀스 작업 하나로 해결할 수 있는 문제는 제한적이기 때문에 여러 맵리듀스 작업을 연결해 워크플로(workflow)로 구성하여 사용됨

• 즉, 맵리듀스 작업 하나의 출력이 다른 맵리듀스 작업의 입력으로 사용되는 방식

• 맵리듀스 프레임워크가 워크플로를 제공하지는 않기 때문에, 맵리듀스 작업은 디렉터리를 기준으로 암묵적으로 연결됨(즉, 두 작업은 완전히 독립적으로 구성)

• 다시말해 일반적인 유닉스 명령 파이프라인(소량의 메모리 버퍼를 사용해 한 프로세스의 출력이 다른 프로세스의 입력으로 전달되는 방식)과는 다른 방식이며, 각 명령의 출력을 임시 파일에 쓰고 다음 명령이 그 임시 파일을 읽는 방식에 가까움(이러한 설계의 장단점에 대해서는 추후, "중간 상태 구체화"에서 논의)

• 이러한 워크플로의 의존성 및 상태를 관리하기 위해 다양한 스케줄러가 개발됨. Oozie, Azkaban, Airflow 등

• Pig, Hive Cascading, Crunch 등 다양한 하둡용 고수준 도구는 다중 맵리듀스를 서로 적절하게 자동으로 엮어서 워크플로를 설정함

리듀스 사이드 조인과 그룹화

• 여러 데이터셋에서 한 레코드가 다른 레코드와 연관이 있는 것은 일반적임. 관계형 모델에서는 외래키(foreign key), 문서 모델에서는 문서 참조(document reference), 그래프 모델에서는 간선(edge)라고 부름

• 데이터베이스에서는 일반적으로 조인 질의를 포함하면 색인을 이용해 관심 있는 레코드를 찾지만, 맵리듀스에는 일반적인 색인 개념이 없음. 따라서 맵리듀스에서는 전체 테이블 스캔을 하게됨. 이는 비용이 많이 드는 것이라고 생각할 수 있지만, 분석 질의는 대량의 레코드를 대상으로 집계 연산을 하는 것이 일반적이라는 가정을 하고 있고, 맵리듀스는 이에 특화된 처리를 하고자 설계됨

• 위 그림과 같은 예제가 있다고 가정하자.

• 가장 간단하게 구현하는 방법은 하나의 활동 이벤트를 훑으면서 나오는 모든 사용자 ID마다 원격 서버에 있는 사용자 데이터베이스에 질의를 보내는 것

• 일괄 처리에서 처리량을 높이기 위해서는 가능한 한 한 장비 내에서 연산을 수행해야함(맵리듀스의 처리과정을 살펴보면 이해할 수 있음)

• 네트워크를 통해 임의 접근 요청을 하는 것은 너무 느리며, 더욱이 원격 데이터베이스에 질의한다는 건 일괄 처리가 비결정적일 수 있다는 의미(원경의 데이터가 변경될 수 있기 때문에)

• 따라서 더 좋은 방법은 사용자 데이터베이스의 사본을 가져와 사용자 활동 이벤트 로그가 저장된 분산 파일 시스템에 넣는 것임. 그렇게 되며 같은 HDFS 상에서 맵리듀스를 사용해 연관된 레코드끼리 모두 같은 장소로 모아 효율적인 처리가 가능

정렬 병합 조인

• 매퍼는 입력 레코드로부터 키와 값을 추출하는 것이 목적이며, 위 그림에서는 사용자 ID임. 한 매퍼는 활동 이벤트를 훑어 사용자 ID를 키로 활동 이벤트를 값으로 추출하며, 다른 매퍼는 사용자 데이터베이스를 훑어 사용자 ID를 키로 사용자 생일을 값으로 추출함

• 맵리듀스 프레임워크에서 키로 매퍼의 출력을 파티셔닝해 키-값 쌍으로 정렬한다면, 같은 사용자의 활동 이벤트와 사용자 레코드는 리듀서의 입력으로 서로 인접해서 들어가게 됨

• 이때 리듀서가 항상 사용자 데이터베이스를 먼저 보고 활동 이벤트를 시간 순으로 보게 하는 식으로 맵리듀스에서 작업 레코드를 재배열하도록 할 수 있는 데, 이를 보조 정렬(secondary sort)라고 함

• 따라서 보조 정렬 후에 리듀서 함수는 모든 사용자 ID 당 한 번만 호출되고 보조 정렬 덕분에 첫번째 값은 생년월일 레코드로 예상할 수 있게되어, 지역 변수에 생년월일을 저장하고 그 다음부터는 사용자 ID가 동일한 활동 이벤트를 순회해서 본 URL과 본 사람의 연령을 쌍으로 출력할 수 있음

• 이 방식을 통하면 리듀서는 특정 사용자 ID의 모든 레코드를 한 번에 처리할 수 있으므로 한 번에 사용자 한 명의 레코드만 메모리에 유지하며 네트워크로는 아무 요청을 보낼 필요가 없게됨

• 이러한 알고리즘을 정렬 병합 조인(sort-merge join)이라고 함.(매퍼 출력이 키로 정렬된 후에 리듀서가 조인의 양측의 정렬된 레코드 목록을 병합하기 때문에)

같은 곳으로 연관된 데이터 가져오기

• 맵리듀스 모델은 올바른 장비로 데이터를 모으는 연산의 네트워크 통신과 데이터를 처리하는 애플리케이션 로직을 분리함(전형적인 데이터베이스 사용 유형과는 대조적인 방식)

• 따라서 맵리듀스는 모든 네트워크 통신을 직접 관리하기 때문에 특정 장비가 죽는 것과 같이 부분적인 실패가 일어나더라도 애플리케이션 코드 단에서는 고민할 필요가 없게됨. 맵리듀스는 애플리케이션 로직에 영향이 가지 않도록 실패한 태스크는 확실하게 재시도함

그룹화

• 조인 외에도 SQL의 GROUP BY절과 같은 그룹화는 많이 쓰임

• 맵리듀스에서 그륩화를 구현하는 가장 간단한 방법은 매퍼가 키-값 쌍을 생성할 때 그룹화할 대상을 키로 지정하는 것임

• 파티션 및 정렬 프로세스가 같은 키를 가진 모든 레코드를 같은 리듀서로 모아서 처리하게 되며, 이는 위에서 살펴본 조인의 구현과 상당히 유사함

쏠림 다루기

• 키 하나에 너무 많은 데이터가 있다면, 맵리듀스의 "같은 키를 가지는 모든 레코드를 같은 장소를 모으는" 패턴에 문제가 생길 수 있음

• 예를 들어 소셜 네트워크에서 소수의 유명인사는 팔로워가 수백만 명에 이르기도 하기 때문에 불균형이 초래되며 불균형한 활성 데이터베이스 레코드를 린치핀 객체 또는 핫 키(hot key)라고 표현함

• 만일 유명 인사 한 사람에 관련된 모든 활동을 리듀서 한 개에서 모은다면 상당한 쏠림 현상(핫스팟)이 발생할 것. 이는 모든 리듀서가 가장 느린 하나의 리듀서가 작업이 완료할 때까지 후속 작업을 시작하지 못한 채 기다려야 한다는 의미

• 이를 완화하는 몇가지 알고리즘이 존재

• Pig에서는 쏠린 조인(skewed join) 메서드가 존재. 이 기법은 간단히 말해 핫 키를 여러 리듀서에 퍼뜨려서 처리하는 방식

• Hive에서는 다른 방법을 사용. 핫 키를 테이블 메타데이터에 명시적으로 지정하며, 핫 키와 관련된 레코드를 나머지 키와는 별도 파일에 저장함. 해당 테이블에서 조인할 때는 핫 키를 가지는 레코드는 맵 사이드 조인(map-side join)을 사용해 처리하는 방식

• 핫 키로 레코드를 그룹화하고 집계하는 작업은 두 단계로 구분해서 수행됨. 첫 번째 맵리듀스 단계는 레코드를 임의의 리듀서로 보내며, 각 리듀서는 핫 키 레코드의 일부를 그룹화하고 키별로 집계해 간소화한 값을 출력함. 두 번째 맵 리듀스 작업은 첫 단계 모든 리듀서에서 나온 값을 기별로 모두 결합해 하나의 값으로 만듦