16주차 스터디

[hyunju-song]

파트

Part03. 파생 데이터 10. 일괄 처리

발표자

유재상

발표 기간

7월 12일 오후 10시

마감기한

7월 12일 오후 10시

참고 링크

일정표 : https://clever-bull-d00.notion.site/bd201441d8a44c83a70375001278c850 책 : http://www.yes24.com/Product/Goods/59566585

[hyunju-song]

Part 3. 파생데이터

• 데이터 모델도 다르고 최적화된 접근 양식도 다른 여러 데이터 시스템을, 일관성 있는 하나의 애플리케이션 아키텍처로 통합하는 문제를 검토

레코드 시스템과 파생 데이터 시스템

• 레코드 시스템
  • 믿을 수 있는 데이터 버전을 저장
  • 레코드 시스템과 다른 시스템 간에 차이가 난다면 정의에 따라 레코드 시스템이 옳다
• 파생 데이터 시스템
  • 데이터는 다른 시스템에 존재하는 데이터를 가져와 특정 방식으로 변환하고 처리한 결과다
  • 기존 데이터를 복제한다는 의미에서 중복이다
  • 읽기 질의 성능 상승
  • 비정규화 과정에서 생성

10장. 일괄 처리

• 신뢰할 수 있고 확장 가능하며 유지보수하기 쉬운 애플리케이션을 구축하는데 중요한 구성요소

유닉스 도구로 일괄처리하기

• 수 기가 바이트의 로그 파일을 수 초 내로 처리할 수 있고, 필요에 따라 분석 방법을 수정하기도 쉽다.
• 유닉스에서의 인터페이스는 파일이다
• 표준 입력과 표준 출력을 사용한다
• 진행상황을 투명하게 파악하기가 쉽다
• 가장 큰 제약은 단일 장비에서만 실행

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

• 유닉스 도구와 비슷한 면이 있지만 수천 대의 장비로 분산해서 실행이 가능하다
• 분산 파일 시스템 상의 파일을 입력과 출력으로 사용한다. → HDFS
• 맵리듀스는 분산 파일 시스템 위에서 대용량 데이터셋을 처리하는 코드를 작성하는 프로그래밍 프레임워크다
• 병렬로 수행한느 코드르 ㄹ직접 작성하지 않고도 여러 장비에서 동시에 처리가 가능

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

• 질의가 조인을 포함하면 여러 색인을 확인해야 하지만 맵리듀스에느 적어도 일반적으로 이야기하는 색인 개념이 없다
• 일괄처리에서 처리량을 높이기 위해서는 가능한 한 한 장비 내에서 연산을 수행해야 한다.
• 같은 곳으로 관련 데이터를 모으는 일종의 그룹화 작업도 중요하다
• 이때 키 하나에 너무 많은 데이터가 연관된다면 문제 발생
  • 불균형한 활성 데이터베이스 레코드를 린치핀 객체 또는 핫키라 한다.
• 조인 로직을 리듀서에서 수행하는 것이 기본
  • 매퍼는 각 입력 레코드에서 키와 값을 추출해 추출한 키-값쌍을 리듀서 파티션으로 할당하고 키별로 정렬한다
  • 입력 데이터에 대한 특정 가정이 필요없다

맵사이드 조인

• 작은 데이터셋과 매우 큰 데이터 셋을 조인하는 경우 가장 간단하게 적용
  • 작은 데이터셋은 전체를 각 매퍼 메모리에 적재 가능할 정도로 충분히 작아야 한다
  • 브로드 캐스트 해시 조인이라 한다.
• 맵사이드 조인의 입력을 파티셔닝한다면 해시조인접근법을 각 파티션에 독립적으로 적용

일괄 처리 워크플로의 출력

• 트랜잭션 처리에 적합하지도 않지만 일괄처리는 분석에도 적합하지 않다.
• 초기에 구글에서 검색엔진에 사용할 색인을 구축하기 위해 처음 사용됐다.
  • 매퍼는 필요에 따라 문서 집합을 파티셔닝하고 각 리듀서가 해당 파티션에 대한 색인을 구축한다
• 검색 색인 외에도 분류기 같은 머신러닝시스템을 구축하거나 추천 시스템을 구축할 수도 있다.

하둡과 분산 데이터베이스의 비교

• HDFS는 파일 시스템이고 맵리듀스는 유닉스 프로세스다
• 데이터베이스(MPP데이터베이스)
  • 특정 모델을 따라 데이터 구조화
  • 데이터를 가져오기 전에 신중하게 모델링
  • 양질의 데이터를 가지게 됨
  • 현실에서는 이상적이기보다 데이터를 빨리 사용가능하게 하는 것이 중요하다
  • 일체식 구조
  • sql질의 언어로 접근이 가능하다
  • 한 장비만 죽어도 대부분은 전체 질의가 중단된다
• 파일 시스템
  • 어떤 데이터 모델과 인코딩을 사용해서도 기록할 수 있다.
  • ETL 프로세스에서 수집은 파일시스템으로, 이후 데이터모델링 진행
  • sql 질의로 처리하기 어려운 이미지 분석등의 데이터 처리 가능
  • 일부 작업의 실패가 전체 작업에 영향을 미치지 않는다 → 내결함성
  • 대용량 작업에 적합

맵리듀스를 넘어

• 직접적으로 맵리듀스를 사용하는 일은 어렵기 때문에 맵리듀스 상에서 추상화된 다양하 ㄴ고수준 프로그래밍 모델이 등장했다.
• 분산 파일 시스템상에 있는 파일들은 중간상태인데, 이는 단순히 한 작업에서 다른 작업으로 데이터를 옮기는 수단이다.
• 중간상태를 파일로 기록하는 과정을 구체화라한다.

데이터플로 엔진

• 맵리듀스처럼 단일 스레드에서 사용자 정의함수를 반복 호출해 한번에 레코드 한개씩 처리한다.
• 입력을 파티셔닝해 병렬화한다.
• 연산자는 맵과 리듀스를 일반화한것.
• 중간 상태를 모두 구체화 할때 생기는 내결함 성이라는 이점이 발생한다
• 태스크가 실패하더라도 다른 장비에서 태스크를 재실행할 수 있고 파일 시스템으로부터 동일한 입력을 다시 읽을 수 있다.
• 복구를 위해 재 계산이 가능하려면 주어진 데이터가 어떻게 연산되는지 추적해야한다.

프리글 처리 모델

• 한 정점이 다른 정점으로 메시지를 보낼 수 있는, 그래프에서 간선을 따라 보내짐
• 액터 모델과 비슷하다.
• 각 정점을 액터로 본다면 정점 상태를 제외하고 정점 사이의 메시지는 내결함성과 지속성이 있다.
• 네트워크 상의 문제로 메시지가 사라지거나 중복되거나 지연되더라도 구현상 다음 반복에서 메시지는 목적지 정점에서 정확히 한번만 처리된다.
• 반복이 끝나는 시점에 모든 정점의 상태를 주기적으로 체크포인트로 저장함으로써 보장된다

고수준 API와 언어

• 직접 맵리듀스 작업을 작성하는 일은 어렵다.

정리

• 중요한 문제
  • 파티셔닝
  • 내결함성
• 조인 알고리즘
  • 정렬 병합 조인
    • 하나의 리듀서에서 호출됨
    • 리듀서 함수에서 병합된 레코드 출력
  • 브로드 캐스트 해시 조인
    • 파티셔닝 없이 작은 테이블을 해시 테이블에 모두 적재
    • 큰입력에서 한번에 하나씩 레코드를 스캔
  • 파티션 해시 조인
    • 같은 방식으로 파티셔닝 하면 해시 테이블 방식을 각 파티션 별로 독립적으로 사용

[KilJaeeun]

part3. 파생 데이터

• 데이터 모델도 다르고 최적화된 접근 양식도 다른 여러 데이터 시스템을 일관된 하나의 애플리케이션 아키텍처로 통합하는 문제에 대해 검토하자.

  레코드 시스템과 데이터 시스템

• 데이터를 저장하고 처리하는 시스템은 크게 두가지 분류인데,
• 레코드 시스템
  • 레코드 시스템은 믿을 수 있는 데이터 버전을 저장하는 ssot 이다. 정규화로 정확하게 한번 표현된다. 다른 시스템과 차이가 있다면, 레코드가 맞음.
• 파생 데이터 시스템
  • 데이터를 가져와 가공한 결과물. 다시 생성 가능.
• 구분해 놓으면 시스템 전체 데이터 플로우가 명확해진다. 시스템 아키텍처를 망치지 않고, 명료성을 갖추기 위해서는 데이터가 어떤 데이터로부터 파생됐는지를 명확히 해야한다.
• 이번 10장은 맵리듀스 oo..

10장 . 일괄 처리

• 시스템은 3가지 유형이 있는데,
• 온라인 시스템
  • 지표 : 응답 시간, 가용성
  • 특: 응답이 올때까지 기다림
• 일괄처리 시스템(오프라인 시스템)
  • 지표 : 처리량
  • 특: 매우 큰 데이터 일괄 처리. 사용자가 끝을 기다리지 않음
• 스트림 처리 시스템(준 실시간 처리)
  • 응답을 기다리지 않고, 입력 데이터르르 소비해 출력데이터를 생산한다.

유닉스로 일괄 처리

• awk, sed, grep, sort, uniq, xargs
• 병렬처리 지원헤서 어지간한 프로그래밍언어보다 나음. 대규모 데이터셋 동작. 유닉스 시스템 내부 서로 호환가능한 인터페이스인 "파일"
• stdin, stout 간 파이프
• 단점 : 단일 장비

  맵리듀스

• 수천대 장비로 분산
• hdfs 설명 : raid 와 리드 솔로몬 코드 삭제 코딩 방식?! 이 헷깔리긴함. 데이터를 조각내 따로 저장으로 이해.
• 매퍼와 리듀스가 있고, 지역성으로 네트워크 비용 절약
• 셔플 : 리듀서 기준으로 파티셔닝하고 정렬한 뒤, 매퍼로부터 데이터 파티션을 복사하는 과정. 정렬된 순서를 유지함.
• 파일 단위 저장

조인의 종류

맵사이드 조인

• 맵 사이드 조인은 작은 크기의 데이터 세트를 메모리에 로드하여 조인 작업을 수행하므로 전체 데이터를 병합하는 리듀스 단계를 거치지 않아도 됩니다. 이를 통해 조인 작업의 성능을 향상시킬 수 있습니다. 그러나 맵 사이드 데이터 세트를 메모리에 로드하기 위해 충분한 메모리 공간이 필요하며, 맵 사이드 데이터 세트의 크기가 매우 클 경우에는 이점이 제한될 수 있습니다.
• 입력 데이터에 대해 특정 가능이 가능해야함.

맵사이드 병합조인

1. 조인할 두 데이터 세트를 각각 정렬합니다. 정렬 작업은 일반적으로 맵리듀스 프레임워크의 맵 단계 이전에 수행됩니다. 이를 위해 데이터를 메모리에 로드하여 정렬합니다.

2. 작은 데이터 세트를 메모리에 로드합니다. 이를 맵사이드 데이터 세트라고 합니다.

3. 맵사이드 데이터 세트의 각 레코드를 맵 함수에 전달하여 조인 작업을 수행합니다. 맵 함수는 맵사이드 데이터 세트의 키를 기반으로 조인을 수행하고, 나머지 데이터 세트를 스캔하여 동일한 키를 가지는 레코드를 찾습니다.

4. 조인된 결과를 맵 단계에서 바로 출력합니다. 리듀스 단계를 거치지 않고 조인 결과를 바로 반환합니다.

리듀스 사이드 조인

• 비정규화 ~~ 열심열심
• 리듀스 사이드 조인은 두 개의 데이터 세트를 조인하는 과정에서 데이터를 맵리듀스 작업의 맵 단계에서 처리합니다. 먼저, 각 데이터 세트에 대해 별도의 맵 작업을 수행하여 데이터를 키-값 쌍으로 변환합니다. 그런 다음, 동일한 키를 갖는 데이터가 같은 리듀스 태스크로 전송됩니다. 리듀스 단계에서는 동일한 키를 갖는 데이터를 취합하고, 이를 기반으로 조인 작업을 수행합니다.
• 장점 : 입력 데이ㅓ 가정 필요없다.
• 단점 : 정렬후 리듀서로 복사한 뒤 입력 병합 과정에 비용이 쎄개게 든다.

브로드캐스트 해시 조인

• 귀여운 사전을 다 붙이기.
• 작은 입력을 큰 입력의 모든 파티션에 브로드 캐스트

파티션 해시 조인

1. 조인할 두 데이터 세트를 동일한 해시 함수를 사용하여 해시 파티션으로 분할합니다. 각 데이터 세트의 레코드는 해시 함수에 적용되어 고유한 해시 키를 생성합니다.

2. 분할된 파티션 간에 조인 작업을 수행합니다. 동일한 해시 키를 갖는 레코드들끼리 조인됩니다. 이때, 동일한 해시 키를 가지는 레코드가 동일한 노드 또는 처리 단위에 위치하므로 네트워크 통신이 최소화됩니다.

3. 조인된 결과를 반환합니다. 각 파티션에서 조인된 결과를 생성하고, 이를 병합하여 최종 조인 결과를 얻습니다.

   정렬 병합 조인

4. 조인 대상 데이터 세트를 각각 정렬합니다. 일반적으로 정렬 작업은 데이터를 메모리에 로드한 후에 수행됩니다.

5. 정렬된 데이터를 병합합니다. 두 데이터 세트에서 동일한 키 값을 갖는 레코드를 찾아 조인합니다. 이때, 정렬된 상태에서 병합 작업을 수행하므로 효율적으로 조인할 수 있습니다.

3, 조인 결과를 반환합니다. 동일한 키 값을 갖는 레코드가 조인되어 새로운 레코드가 생성됩니다. 이러한 결과는 일반적으로 새로운 데이터 세트로 생성되거나 원하는 형식으로 출력됩니다.

skew problem

• 핫스팟 : 쏠린키, 인플루언서
• 핫스팟 완화 알고리즘
• Pig 에 skew join 은 어떤 키가 핫키인지 샘플링하고, 핫키를 여러 리듀서에 퍼뜨려 처리하게 한다. 복제비용이 들어도, 병렬 효과가 크다.
• hive 에서 핫키는 테이블 메타데이터에 명시적으로 지정하고, 핫키와 관련된 레코드를 나머지 키와 별도 파일에 저장한다. 해당 테이블에서 조인할때, 맵사이드 조인으로 처리.

다른 시스템들 소개

• sql 엔진을 붙이는 경우도 있고
• 파일 저장이 힘드니, 중간상태를 추상화한 스파크 라던지 테즈 , 플링크 등이 있다. 이들은 전체 워크플로를 독립된 하위 작업으로 나누지 않고 작업하나로 다룬다. 데이터 플로 엔진 이라 부름 (dag). 일괄 처리 그래프 최적화 방식으로 프리글 처리 모델 등의 그래프로 작업을 수행 처리 방법론에 대해 이야기. "메시지 전달"로 매퍼가 특정 리듀서를 호출하는게 대기 시간을 줄인다고함. 질의 최적화기(sql 을 최적화한다는 생각. 그래프와는 또 별개) 에 대한 언급.

[JSYoo5B]

https://docs.google.com/presentation/d/1-udTkUIy_ulUagFLLQKUBbUAzbh6GMoRf6CdeK3li5w/edit?usp=sharing

[jasonkang14]

Batch Processing

• request ↔ response / query ↔ result
  • 무언가를 달라고 요청하면 잠시 후에 원하는 것을 돌려주는 방식
  • 사용자가 너무 오래 기다리지 않도록 response time 이 중요함
• 시스템에는 3가지 방식이 있음
  • services - online systems
    • 서비스는 클라이언트의 요청을 기다림
    • 요청을 받으면, 해당 요청을 최대한 빠르게 처리해서 응답하려고 함
    • response time이 서비스의 성능을 측정하는 주요 지표이고
    • availability가 매우 중요함
  • batch processing - offline systems
    • 많은 양의 인풋 데이터를 활용해서, 해당 데이터를 처리하기 위해 job을 실행함
    • job 은 일반적으로 오래걸리기 때문에, 사용자는 job이 끝날때까지 기다리지 않음
    • 일반적으로 batch job들은 주기적으로 실행되도록 스케쥴됨
    • batch의 성능은 특정 인풋 사이즈를 처리하는데 얼마나 오래 걸리는지를 측정하는 throughput으로 결정됨
  • stream processing - near-real-time systems
    • online과 offline사이의 그 어딘가에 존재함
    • batch처럼 거대한 데이터를 처리하는 경우도 있지만, stream은 batch와 다르게 이벤트가 발생하면 그 직후에 작동함
    • 따라서 batch보다 빠른편임

Batch Processing with Unix Tools

• nginx와 유사한 append only log

Simple Log Analysis

• nginx log를 활용해서 가장 많이 접근하는 페이지를 파악하는 방법

  cat /var/log/nginx/access.log |
    aws '{print $7}' |
    sort |
    uniq -c |
    sort -r -n |
    head -n 5

  • log파일을 읽어오고
  • 모든 줄을 공백을 기준으로 나누고. url에 해당하는 7번째 필드를 추출함
  • url을 알파벳 순으로 정렬하고
  • unique한 값들만 남기고, 몇 번 일어났는지 카운터를 추가
  • 위에 카운터대로 정렬을 한 번 더 실행
  • 맨 위에 5개 리턴
• 꽤 큰 용량의 로그를 몇 초 안에 분석할 수 있음

Chain of commands versus custom program

• unix 명령어를 여러개 이어서 사용하지 않고, 프로그램을 작성할수도 있음
• 프로그래밍 언어를 사용해서 작성하는 경우 큰 차이가 없어 보이지만, 용량이 큰 파일을 대상으로 작동시키는 경우 큰 차이가 있음

Sorting versus in-memory aggregation

• Ruby 예제는 url들의 hash table을 메모리에 저장함
  • 각각의 url들이 몇 번 호출됐는지 매핑되어 있음
• Unix pipeline은 hash table을 사용하지 않고, sorting 알고리즘을 활용함
• 용량이 작다면 메모리에 충분히 저장할 수 있지만, 로그파일의 용량이 메모리보다 크다면 불가능함

The Unix Philosophy

• 각각의 프로그램은 한가지 일을 잘 처리한다. 새로운 일을 처리하고 싶다면 기존 프로그램에 기능을 추가하지 말고 새로 만들어라
• 모든 프로그램의 output은 다른 프로그램의 input이 될 수 있다.
• 가급적 빨리 새로운 프로그램을 실행하고, 불필요한 부분은 버려라
• 프로그래밍 태스크를 줄이기 위해 다양한 도구를 사용해라

A uniform interface

• 프로그램의 아웃풋이 다른 프로그램의 인풋으로 사용되려면, 같은 데이터 포맷을 사용해야 한다
  • Unix에서는 파일을 사용한다. 파일은 사실상 순서대로 정렬된 byte sequence이기 때문
  • 해당 파일들을 ASCII text처럼 처리하는 경우도 많음
  • parsing하기 위해서 빈칸, 탭, 컴마 등을 사용함
• 하지만 가독성이 떨어지는 경우도 많음

Separation of logic and writing

• 프로그램이 아무것도 특정하지 않으면 stdin은 키보드에서 받아서 stdout은 화면으로 전송됨
  • 인풋을 파일로부터 받거나, 아웃풋을 파일로 보낼수도 있긴함
  • 그리고 아웃풋을 다른 프로그램의 인풋으로 바로 보낼수도 있음
• 파일을 바로 읽어들여도 되지만, Unix는 stdin과 stdout의 사용을 지향함
  • 특정 file이나 filepath에 국한되지 않고 프로그래밍이 가능하기 때문
  • 하지만 인풋이나 아웃풋이 복잡하거나 하나가 아니라면 좀 복잡해짐

Transparency and experimentation

• Unix tool을 사용하면 프로그램이 어떻게 동작하는지 파악하기 쉽다
  • input file들은 immutable하고
  • pipeline은 언제든지 멈출 수 있고
    • 중간에 멈추면 less로 아웃풋이 나감
  • output으로 생성된 파일을 다른 프로그램의 input으로 사용 가능
    • 전체를 재시작 하지 않고, 중간부터 다시 구동할 수 있음

MapReduce and Distributed Filesystems

• Unix와 유사한데, 여러 기계들에 분산되어 작업을 처리할 수 있다
  • side effect 없이 input → output이라는 특성이 매우 유사하다
  • output은 주로 HDFS의 파일로 write된다
  • HDFS는 해당 output을 분산시켜서 저장하는 느낌

MapReduce Job Execution

• MapReduce는 기본적으로 큰 규모의 데이터를 처리할 수 있음
  • break files → map → sort → reduce 의 순서
  • mapper는 인풋을 받으면 원하는 key-value pair를 생성하는 역할이고
  • reducer는 sorting된 해당 key-value pair를 분석/처리하는 역할임

Distributed execution of MapReduce

• Unix는 한번에 하나씩만 가능하지만 MapReduce는 병렬로 작업할 수 있음
  • 병렬성을 명시적으로 처리하는 코드를 작성할 필요가 없음
  • Mapper와 Reducer는 한 번에 하나의 레코드만 처리함
    • 입력이 어디서 오는지 또는 출력이 어디로 가는지 알 필요가 없음
    • 프레임워크가 기기 간 데이터 이동의 복잡성을 처리할 수 있게함
• Hadoop MapReduce 작업의 데이터 흐름은 파티셔닝에 기반한 병렬화를 사용함
  • 일반적으로 작업의 input은 HDFS의 디렉토리,
  • input 디렉토리 내의 각 파일 또는 파일 블록은 별도의 Map 작업으로 처리될 수 있는 별도의 파티션으로 간주함
  • MapReduce 스케줄러는 입력 파일의 복제본을 저장하는 기기 중 하나에서 각 Mapper를 실행함
  • Reducer 부분의 계산도 partition됨.
    • 모든 key-value pairrk 동일한 Reducer에서 끝나도록 하기 위해 key의 해시를 사용하여 특정 key-value pair를 받아야 하는 Reduce 작업을 결정
• 분산처리 시 다른 mapper들이 같은 Key로 key-value pair를 생성하면 sort할 때 가까이로 이동함

MapReduce workflows

• 단일 MapReduce 작업으로 해결할 수 있는 문제의 범위는 제한적
  • 단일 MapReduce 작업으로는 URL 별 페이지 뷰 수를 결정할 수 있는데,
  • 같은걸로 가장 인기 있는 URL을 결정할 수는 없음
• 따라서 MapReduce 작업들이 워크플로우로 연결되고 하나의 output 이 다음 작업의 input이 되는 것
  • Hadoop MapReduce 프레임워크는 워크플로우에 대한 특별한 지원이 없으므로, 이 연결은 디렉토리 이름에 의해 암시적으로 이루어짐
  • 첫 번째 작업은 출력을 HDFS의 지정된 디렉토리에 작성하도록 설정되어야 하며, 두 번째 작업은 그 같은 디렉토리 이름을 입력으로 읽도록 설정
  • 체인화된 MapReduce 작업은 유닉스 명령어의 파이프라인과는 달리, 각 명령어의 출력이 임시 파일에 작성되고, 다음 명령어가 임시 파일을 읽는 명령어 시퀀스와 더 유사함.
• 배치 작업의 출력은 작업이 성공적으로 완료되었을 때만 유효
  • 따라서, 워크플로우에서 하나의 작업은 그 입력 디렉토리를 생성하는 이전 작업들이 성공적으로 완료되었을 때만 시작할 수 있
  • 이러한 작업 실행 간 의존성을 처리하기 위해, Oozie, Azkaban, Luigi, Airflow, Pinball 등 다양한 Hadoop 워크플로우 스케줄러를 사용함

Reduce-Side Joins and Grouping

• 많은 데이터셋에서 하나의 레코드가 다른 레코드와 연관성을 갖는 것이 일반적.
  • 관계형 모델의 foreign key, 문서 모델의 문서 참조, 그래프 모델의 edge등.
  • 연관성의 양쪽 레코드에 접근이 필요한 코드가 있는 경우 언제든지 조인이 필요
• 데이터베이스에서 소수의 레코드만을 관여시키는 쿼리를 실행하면,
  • 데이터베이스는 일반적으로 인덱스를 사용하여 관심 있는 레코드를 빠르게 찾을 수 있음
  • 조인이 포함된 쿼리는 여러 개의 인덱스 조회를 필요로 할 수 있음.
  • 그러나, MapReduce에는 일반적인 의미에서의 인덱스 개념이 없음
• MapReduce 작업이 일련의 파일을 입력으로 주어지면, 이는 그 파일들의 전체 내용을 read.
  • 데이터베이스에서 '전체 테이블 스캔' 과 유사한 느낌
  • 소수의 레코드만 읽고 싶은 경우, 전체 테이블 스캔은 인덱스 조회에 비해 엄청난 비용이 발생
  • 그러나, 분석 쿼리에서는 대량의 레코드에 대해 집계를 계산하는 것이 일반적임
  • 처리를 여러 기계에 병렬화할 수 있다면 전체 입력을 스캔하는 것이 합리적일 수 있음

sort-merge joins

• Mapper의 목적은 각 input record에서 key-value를 추출하는 것
• Mapper output을 key에 의해 파티션하고 key-value pair를 sort하면,
  • 같은 사용자 ID를 가진 모든 활동 이벤트와 사용자 레코드가 reducer input 시 서로 인접해집니다.
  • secondary sort를 활용하면 reducer는 사용자 데이터베이스에서의 레코드를 먼저 보고, 그 다음에 타임스탬프 순서로 활동 이벤트를 볼 수 있음
  • 이후 reducer는 실제 조인 로직을 쉽게 수행할 수 있음
    • reducer 함수는 각 사용자 ID에 대해 한 번씩 호출되며, secondary sort 덕분에 첫 번째 값은 사용자 데이터베이스에서의 생년월일 레코드가 됨
    • reducer는 생년월일을 로컬 변수에 저장한 다음, 같은 사용자 ID를 가진 활동 이벤트를 반복하여 조회된 URL과 조회자의 나이(년) 쌍을 출력함
    • 후속 MapReduce 작업은 각 URL에 대한 조회자 연령 분포를 계산하고, 연령 그룹별로 클러스터링 할 수 있음
• reducer는 특정 사용자 ID에 대한 모든 레코드를 한 번에 처리
  • 한 번에 하나의 사용자 레코드만 메모리에 유지하면 되고
  • 네트워크에 요청을 보낼 필요가 없음
  • 이 과정을 sort-merge join이라고 부름
    • mapper output이 key에 따라 sort되고, reducer가 조인의 양쪽에서 정렬된 레코드 목록을 join하기 때문

Bringing related data together in the same place

• sort-merge join에서는 mapper → sorting 하면서 join에 필요한 데이터들이 reducer input에 모이도록 함
  • 데이터는 이미 sort되어있기 때문에 reducer는 낮은 메모리 오버헤드로 레코드를 처리할 수 있음
• mapper가 reducer에게 message를 보내는 느낌
  • mapper의 output이되는 key-value pair에서, key가 목적지(reducer)의 주소가 됨

GROUP BY

• MapReduce로 에서는 mapper를 설정하여 key-value pair를 원하는 그룹화 key를 사용하도록 하면됨
  • 같은 key를 가진 값들을 파티셔닝해서 모든 레코드를 동일한 reducer로 모음
  • group by와 join은 MapReduce 위에서 구현될 때 매우 유사
• 특정 사용자 세션의 모든 활동 이벤트를 정리하여 사용자가 취한 행동의 순서를 파악할 수 있음(sessionization)
  • A/B 테스트 등 마케팅에서 많이 쓰임

Handling skew

• "동일한 키를 가진 모든 레코드를 같은 곳에 모으는" 패턴은 단일 key에 관련된 데이터가 매우 많은 경우에 문제가 발생함
  • 인스타에서 팔로워 엄청 많은 유명인들은 hot key가 되어버림
  • key에 따라 reducer작업이 이루어지니 hot key들이 hot spot이 되어버려 엄청 오래걸림
  • 후속작업은 모든 reducer의 작업을 기다려야하니 전체 작업 소요 시간도 오래걸림
• hot key를 다양한 알고리즘을 사용해서 보완할 수 있음
  • Pig는 샘플링 작업을 실행하여 어떤 키가 핫 키인지 결정함
  • 실제 조인을 수행할 때, mapper는 핫 키와 관련된 모든 레코드를 무작위로 선택된 여러 reducer 중 하나에게 전송
  • hot key와 관련된 작업들을 모든 reducer에 복사해줘야함
  • 병렬화 측면에서는 좋은데 Join된 값들을 여러 reducer에 복제해줘야하는 문제가 있음
• Hive는 hot key를 metadata에 작성함
  • 해당 key와 관련된 record들을 별도 파일에 저장
  • hot key에 대해서는 map-side join을 실시

Map-Side Joins

• reduce-side join에서 mapper는 reducer를 돌리기 위해서 준비하는 역할을 함
  • mapper가 reducer가 데이터를 join할 수 있도록 정렬까지 해줌
  • 그러나 reducer에서 복사, join 할 때 많은 비용이 발생할 수 있음
    • 특히 메모리 버퍼 용량이 부족하면 디스크에 여러번 write될 수 있음
• map-side join은 이에비해 훨씬 빠름
  • reducer가 필요없고
• mapper가 HDFS에서 Input file block을 Read한 후에 output file에 write함

Broadcast hash joins

• map-side join은 큰 데이터셋과 작은 데이터셋을 join할 때 제일 간단함
  • 작은 데이터셋은 mapper에서 메모리에 전부 load할 정도로 작은경우
  • mapper가 작은 데이터셋을 메모리 내 해시 테이블로 read함
  • 사용자 활동 이벤트를 스캔하면서 각 이벤트에 대한 사용자 ID를 해시 테이블에서 찾음
  • map task자체는 여러개일 수 있음.
• 브로드캐스트 라는 단어는 큰 입력의 각 파티션에 대한 mapper 가 작은 입력의 전체를 읽는다는 것
• 해시 라는 단어는 해시 테이블을 사용한다는 것
• 메모리 내 해시테이블 말고 로컬디스크에 read-only index로 저장할 수도 있음
  • 자주 사용되는 부분은 운영체제의 페이지 cache에 남아있음
  • 메모리에 담지 않아도 랜덤 엑세스 접근이 가능함

Partitioned hash joins

• hash join을 각 파티션에 독립적으로 적용함
  • 사용자 id를 마지막 소수점 기준으로 파티셔닝 하고
  • 파티션별로 join을 하는 방식
  • 파티션 되어있으니 메모리에 올라가는 데이터가 더 적다는 장점이 있음

Map-side merge joins

• 동일한 key를 기준으로 정렬될 경우 사용 가능
  • input이 메모리에 적합한 크기인지 여부는 중요하지 않음
  • mapper가 reducer에서 수행되는 merge를 처리할 수 있기 때문

MapReduce workflows with map-side joins

• map-side join / reduce-side join은 output의 구조에 영향을 주게됨
  • reduce-side join의 output은 key로 파티션/정렬 되지만
  • map-side join의 output은 더 큰 input과 동일한 방식으로 파티션/정렬 됨
• map-side join은 input 데이터셋의 크기, 정렬, 파티셔닝에 대해 더 많은것을 가정하고 들어감
  • 분산 파일시스템에서 데이터셋의 물리적 구조에 대해 알아야만 조인 전략을 최적화할 수 있음
  • 데이터가 저장된 디렉토리의 이름과 인코딩 형식만 알면 충분하지 않고,
  • 데이터가 어떤 키로 파티션화되고 정렬되었는지 그리고 파티션의 수를 알아야 함

The Output of Batch Workflows

• 데이터 처리가 끝난 후 결과는?
  • OLTP 쿼리는 일반적으로 인덱스를 사용하여 키를 통해 소수의 레코드를 조회함
  • 반면에 분석 쿼리는 대개 많은 수의 레코드를 스캔하고 그룹화하고 집계함
• batch는 트랜잭션 처리도 아니고 분석도 아님.
  • batch 처리는 일반적으로 입력 데이터셋의 큰 부분을 스캔하기 때문에 분석에 더 가깝긴함
  • 그러나, MapReduce workflow는 분석 목적으로 사용되는 SQL 쿼리와 다름

Building search indexes

• Google의 MapReduce 원래 사용처는 검색 엔진을 위한 인덱스를 구축하는 것
  • 5~10개의 MapReduce 작업의 워크플로우로 구현
  • 이제는 아닌데 그래도 인덱스 설정할 때 참고하면 좋긴함
• MapReduce를 활용하면 병렬적으로 인덱스 생성/처리가 가능함
  • mapper가 문서 집합을 파티션하고
  • reducer가 파티션별로 인덱스를 구축하고
  • 인덱스는 HDFS에 기록하는 방식을 따름
• 인덱스 파일은 수정할 수 없고, 수정이 필요한 경우 새로 생성해서 교체해야함

Key-value stores as batch process output

• 검색 인덱스는 batch의 output중 하나임
  • 머신러닝 학습 시스템을 구축하는데도 많이 사용함
  • output은 종종 어떤 종류의 데이터베이스임
• 이러한 데이터베이스는 hadoop 인프라와 별개로 사용자 요청을 처리하는 웹 애플리케이션에서 쿼리하는 경우가 많음
  • mapper나 reducer에서 원하는 데이터베이스나 클라이언트 라이브러리를 사용할 수 있음
  • 그리고 batch작업에서 데이터베이스 서버에 레코드를 작성함
    • 하지만 문제가 좀 있음
    • 모든 개별 레코드에 대해 네트워크 요청을 하는 것은 배치 작업의 정상적인 처리량보다 느림
    • 모든 mapper나 reducer가 동시에 같은 output 데이터베이스에 write를 시작하면 쿼리 성능이 저하될 가능성
  • 해결방법은
    • batch 작업 내에서 새 데이터베이스를 구축
    • 분산 파일 시스템의 작업 output 디렉토리에 파일로 작성
• mapper를 사용하여 키를 추출하고 그 키로 정렬하는 것은 인덱스를 구축하는 데 필요한 작업
  • 대부분의 key-value store가 read-only이기 때문에 데이터 구조는 상당히 단순함

Philosophy of batch process outputs

• unix 기본개념과 유사하게 input을 받아서 output을 작성함
  • 버그가 발생하면, 이전 버전으로 돌려서 제대로 된 Output을 생성(?)할 수 있음
  • 아니면 새로운 버전별로 output을 관리하면, 이전 버전의 output을 사용해도 됨
    • transaction을 지원하는 데이터베이스는 이런게 불가능함
    • 코드를 롤백해도 데이터는 그대로 에러임
• mapper/reducer 작업이 실패하면 자동으로 re-schedule하고, 동일한 input으로 재시작됨
  • 버그로 인한 에러면 retry몇 번 하다가 멈춤
    • transaction과 유사하게 실패한 job들의 output은 버려짐
  • 일시적인 문제면 다시 시도해서 처리함

Comparing Hadoop to Distributed Databases

• hadoop은 분산 unix느낌

Diversity of storage

• 데이터베이스는 특정 모델에 따라 데이터를 구조화하는 것
  • 분산 파일 시스템 파일들은 어떠한 데이터 모델과 인코딩을 사용하여 작성할 수 있는 바이트 시퀀스
  • 레코드의 컬렉션일 수도 있지만 다양한 데이터 타입으로 저장이 가능함
• 심하게 말하면
  • Hadoop은 무차별적으로 데이터를 HDFS에 던져두고 나중에 어떻게 추가 처리할지 그 때 가서 걱정함
  • 반면에, MPP 데이터베이스는 일반적으로 데이터와 쿼리 패턴을 세심하게 미리 모델링한 후에 데이터를 데이터베이스의 전용 저장 형식으로 가져와야 함
• 순수주의자의 관점에서 보면 세심한 모델링과 가져오기가 바람직해 보일 수 있음
  • 데이터베이스 사용자가 더 나은 품질의 데이터로 작업할 수 있게 하기 때문
  • 하지만 이상적인 데이터 모델을 미리 결정하려는 시도보다는 단순히 데이터를 빨리 사용할 수 있게 하는 것이 좋을 때가 많음
• 데이터 웨어하우스와 비슷함
  • 단순히 큰 조직의 다양한 부분에서 데이터를 한 곳에 모으는 것만으로도 가치가 있음
    • 이전에 분산되어 있던 데이터 세트 간에 조인을 가능하게 하기 때문.
  • MPP 데이터베이스가 요구하는 세심한 스키마 설계는 중앙 집중화된 데이터 수집을 늦춤
• 데이터를 일단 때려박으면 데이터 해석은 다른곳에서 담당하는 느낌임
  • 따라서 Hadoop은 주로 ETL 구현에 많이 사용됨

Diversity of processing models

• MPP 데이터베이스는 디스크 상의 저장 레이아웃, 쿼리 계획, 스케줄링, 실행 등을 처리
  • 각 컴포넌트들은 모두 데이터베이스의 특정 필요성에 대해 튜닝되고 최적화될 수 있음
  • SQL을 사용해서 데이터에 접근할 수 있음
• 하지만 SQL로 모든 것을 처리할 수는 없음
  • 머신러닝이나 추천시스템 등을 구축할 때는 일반적인 데이터 처리 모델이 필요함
• MapReduce를 활용하면 큰 데이터셋을 처리할 수 있음
  • Hive는 HDFS+MapReduce위에 SQL 쿼리 실행 엔진을 구축한 것
  • MapReduce의 한계를 극복하기 위해 Hadoop에는 이것저것 많이 추가됨

Designing for frequent faults

• MPP 데이터베이스와 MapReduce 비교
  • 오류 처리
    • MPP 데이터베이스는 쿼리 실행 중 노드 실패 시 전체 쿼리를 중단하고 사용자가 다시 제출하거나 자동으로 다시 실행
    • MapReduce는 map 또는 reduce 작업의 실패를 전체 작업에 영향을 주지 않고 복구
  • 메모리와 디스크 사용
    • MPP 데이터베이스는 가능한 한 많은 데이터를 메모리에 유지
    • MapReduce는 메모리에 맞지 않을 것이라고 가정하고 데이터를 디스크에 쓰는 것을 선호
• MapReduce 디자인의 배경
  • Google의 혼합 사용 데이터센터에서 온라인 서비스와 오프라인 배치 작업이 동일한 기계에서 실행됨
  • 모든 작업에는 리소스 할당이 있으며, 높은 우선순위의 작업이 더 많은 리소스가 필요한 경우, 동일한 기계에서 우선순위가 낮은 작업을 종료하여 리소스를 확보
  • 이러한 아키텍처는 비생산(우선순위가 낮은) 컴퓨팅 리소스를 과도하게 할당하도록 허용함

Beyond MapReduce

• MapReduce는 분산 시스템을 위한 많은 프로그래밍 모델 중 하나일뿐
  • 데이터의 양, 구조 및 처리 유형에 따라 다른 도구가 계산을 표현하는 데 더 적합할 수 있음
• MapReduce 사용의 어려움
  • 복잡한 처리 작업을 원시 MapReduce API를 사용하여 구현하는 것은 매우 어렵고 번거로움
  • MapReduce를 직접 사용하는 어려움에 대응하여, 다양한 고수준 프로그래밍 모델(Pig, Hive, Cascading, Crunch)이 MapReduce 위의 추상화로서 생성됨

Materialization of Intermediate State

• MapReduce 작업은 다른 작업과 독립적임
  • 분산 파일 시스템상의 input / output 디렉토리로 소통함
  • 한 작업의 output을 다른 작업의 input으로 사용하려면 디렉토리 구조를 동일하게 구성해야함
  • 그리고 첫번째 작업이 완료된 후에만 두번째 작업을 시작할 수 있음
• 첫번째 작업의 output이 다양한 곳에서 사용된다면, 데이터를 글로벌? 느낌으로 저장해서 느슨한 결합을 시도할 수 있음
  • 하지만 해당 output이 같은 팀 내에서만 사용되면 전에 언급한 message queue느낌임

Dataflow engines

• Spark, Tez, Flink 등의 분산 배치 계산을 위한 새로운 실행 엔진들이 개발됨
  • 사용자 정의 함수를 반복적으로 호출하여 하나의 레코드를 단일 스레드에서 처리하는 방식으로 동작
  • MapReduce처럼 엄격한 맵과 리듀스의 역할을 가지지 않고, 보다 유연한 방식으로 조합이 가능함
• 데이터플로 엔진은 MapReduce 모델에 비해 여러 가지 이점을 제공함
  • 다양한 옵션을 통해 연산자의 output을 다른 연산자의 input에 연결 가능
  • MapReduce 워크플로우와 동일한 계산을 구현할 수 있고, 최적화로 인해 실행 속도가 훨씬 빠름
  • 연산자는 맵과 리듀스의 일반화된 형태이므로, Pig, Hive, Cascading으로 구현된 워크플로우를 코드를 수정하지 않고 MapReduce에서 Tez 또는 Spark로 전환할 수 있음

Fault Tolerant

• MapReduce에서는 중간 상태를 완전히 저장함으로써 내구성을 확보하고 장애 허용성을 제공함
  • Spark, Flink 및 Tez는 중간 상태를 HDFS에 기록하지 않음
  • 장애 허용성에 대해 다른 접근 방식을 취해야함
    • 장애 발생 시, 손실된 중간 상태는 사용 가능한 다른 데이터에서 다시 계산됨
• 비결정론적 연산자의 경우, 장애 발생 시 다운스트림 연산자도 중지시키고 새로운 데이터에 대해 다시 실행하는 것이 일반적
  • 연산자를 결정론적으로 만드는 것이 중요하지만, 실수로 비결정론적 동작이 포함될 수 있음
  • 중간 데이터가 충분히 작거나 계산이 매우 CPU 집약적인 경우, 중간 데이터를 저장하는 것이 재계산하는 것보다 비용이 적을 수 있음

Discussion of materialization

• MapReduce는 각 명령의 출력을 임시 파일에 쓰는 것과 유사함
  • 데이터플로 엔진은 Unix 파이프와 더 유사하게 보인다.
  • 플링크는 특히 파이프라인 실행을 중심으로 구축되어 있음
    • 입력이 완료될 때까지 기다리지 않고 즉시 처리를 시작함
• 정렬 작업은 입력 전체를 소비해야만 출력을 생성할 수 있음
  • 일부 연산자는 임시 상태를 누적해야 함

Graphs and Iterative Processing

• Batch processing에서는 그래프 알고리즘을 사용하여 오프라인 처리 및 분석을 수행할 수 있음.
• 그래프 알고리즘은 간선과 vertex를 탐색하고 정보를 전파하는 방식으로 작동함.
• MapReduce는 반복적인 알고리즘에 적합하지 않아서 효율적인 구현이 어려울 수 있음.
• 반복적인 알고리즘은 MapReduce의 단일 패스 특성과는 다르게 전체 입력 데이터를 읽고 완전히 새로운 출력 데이터를 생성함.

The Pregel processing model

• 그래프의 배치 처리를 위한 최적화 방법으로는 대량 동기 병렬(Bulk Synchronous Parallel, BSP) 계산 모델이 인기가 있음.
  • Apache Giraph, Spark의 GraphX API, Flink의 Gelly API 등에서 BSP 모델을 구현함.
  • Pregel 모델이라고도 알려져 있으며, Google의 Pregel 논문에서 그래프 처리를 위한 이 접근 방식을 널리 알리게 되었음.
• Pregel에서는 각 vertex가 다른 vertex으로 "메시지를 보내는" 개념을 사용하며, 일반적으로 이러한 메시지는 그래프의 간선을 따라 전송됨.
  • 각 반복에서는 각 vertex에 대해 함수가 호출되며, 해당 vertex에게 보내진 모든 메시지가 전달됨.
  • Pregel 모델에서는 vertex가 다음 반복에서도 상태를 기억하므로 함수는 새로운 들어오는 메시지만 처리하면 됨.

Falut tolerance

• Pregel 작업의 성능을 향상시키기 위해 vertex들은 직접적인 쿼리가 아닌 메시지 전달을 통해 통신함.
  • 메시지를 일괄로 처리할 수 있으며 통신을 기다릴 필요가 적어짐.
  • 기다림이 발생하는 유일한 시간은 반복 간에 있음.
    • Pregel 모델은 한 번에 전송된 모든 메시지가 다음 반복에서 전달되도록 보장하기 때문에
    • 이전 반복은 완전히 완료되고 모든 메시지가 네트워크를 통해 복사된 후 다음 반복이 시작될 수 있음.
• 메시지는 기반이 되는 네트워크에서 손실, 중복 또는 임의의 지연이 발생할 수 있음
  • Pregel 구현은 메시지가 다음 반복에서 대상 vertex에서 정확히 한 번 처리되도록 보장함.
• Pregel은 내결함성을 투명하게 처리하여 Pregel 위의 알고리즘들을 간소화하는 프로그래밍 모델을 제공함.
  • 이 내결함성은 각 반복의 끝에서 모든 vertex의 상태를 주기적으로 체크포인팅하여 지속성 있는 저장소에 전체 상태를 기록함으로써 달성됨.
  • 노드가 실패하고 메모리 상태가 손실된 경우, 가장 간단한 해결책은 그래프 계산을 마지막 체크포인트로 롤백하고 계산을 재시작하는 것임.
  • 알고리즘이 결정론적이고 메시지가 로깅된 경우, 손실된 파티션만 선택적으로 복구하는 것도 가능함

Parallel execution

• vertex는 어떤 물리적인 머신에서 실행되고 있는지 알 필요가 없음
  • 다른 vertex에 메시지를 보낼 때는 단순히 vertex ID로 보내면 됨.
  • 그래프를 파티션하는 일은 프레임워크에 달려 있음.
    • 프레임워크는 어떤 vertex가 어떤 머신에서 실행되는지,
    • 메시지를 네트워크를 통해 어떻게 라우팅하여 올바른 위치에 도달시킬지 결정함.
• 프로그래밍 모델은 한 번에 하나의 vertex에 대해 작업하기 때문에 프레임워크는 그래프를 임의의 방식으로 파티션할 수 있음.
  • 통신이 많이 필요한 vertex들이 동일한 머신에 공존하도록 파티션되면 좋음,
  • 하지만 이러한 최적의 파티셔닝을 찾는 것은 어려움.
  • 실제로는 그래프는 종종 임의로 할당된 vertex ID에 따라 파티션됨으로써 관련된 vertex들을 함께 묶는 시도를 하지 않음.
• 그래프 알고리즘은 종종 여러 머신 간의 통신 오버헤드가 많이 발생
  • 중간 상태(노드 간에 전송된 메시지)는 원래 그래프보다 크게 될 수 있음.
  • 네트워크를 통해 메시지를 전송하는 오버헤드는 분산 그래프 알고리즘의 성능을 크게 늦출 수 있음.
• 따라서, 그래프가 단일 컴퓨터의 메모리에 들어갈 수 있는 경우,
  • 단일 컴퓨터에서 실행되는 알고리즘이 분산 배치 프로세스보다 성능이 우수할 확률이 높음
  • 그래프가 메모리보다 큰 경우에도, GraphChi와 같은 프레임워크를 사용하여 단일 컴퓨터의 디스크에 그래프가 들어갈 수 있음
  • 그래프가 단일 머신에 맞지 않을 정도로 큰 경우, Pregel과 같은 분산 접근 방식을 피할 수 없음

High-Level APIs and Languages

• MapReduce를 넘어서는 실행 엔진들은 대규모 데이터를 처리하고 저장하는 능력을 향상시킴
  • 고수준 언어와 API들은 MapReduce 작업의 번거로움을 줄여줌
  • Tez의 등장으로 코드를 재작성하지 않고도 새로운 데이터플로우 실행 엔진으로 전환할 수 있음
• 데이터플로우 API는 관계형 스타일의 구성 요소를 사용하여 계산을 표현하며, 내부적으로 다양한 알고리즘을 활용
• 단일 컴퓨터의 메모리에 그래프가 들어갈 수 있다면,
  • 단일 머신에서 처리하는 것이 분산 배치 처리보다 효율적일 수 있음
  • 그래프가 메모리에 들어가지 않더라도, 그래프를 하나의 컴퓨터의 디스크에 저장하고 처리하는 것은 GraphChi와 같은 프레임워크를 사용하여 가능함
  • 그래프가 하나의 머신에 들어갈 수 없을 경우 Pregel과 같은 분산 접근 방식을 사용하여 효율적으로 병렬 그래프 알고리즘을 처리할 수 있음

The move toward declarative query languages

• 조인을 간단하게 할 수 있다면 프레임워크가 조인 입력의 속성을 분석하고 자동으로 어떤 조인 알고리즘이 해당 작업에 가장 적합한지를 결정할 수 있음
  • Hive, Spark 및 Flink는 이를 수행하기 위한 비용 기반의 쿼리 옵티마이저를 가지고 있음
  • 조인의 순서를 변경하여 중간 상태의 양을 최소화할 수도 있음
  • 조인 알고리즘의 선택은 배치 작업의 성능에 큰 영향을 미칠 수 있음
• 그러나 MapReduce 및 데이터플로우 기반의 후속 시스템은 SQL의 쿼리 모델과는 매우 다름
  • MapReduce는 함수 콜백을 기반으로 구축됨
  • 각 레코드 또는 레코드 그룹에 대해 사용자 정의 함수(mapper 또는 reducer)가 호출됨
  • 해당 함수는 출력을 결정하기 위해 임의의 코드를 호출할 수 있음
• 임의의 코드를 쉽게 실행할 수 있는 자유로움은 batch와 MPP를 구분짓는 요소임
  • 데이터베이스는 사용자 정의 함수를 작성할 수 있는 기능을 갖추고 있지만,
    • 사용하기 번거롭고
    • 대부분의 프로그래밍 언어에서 널리 사용되는 패키지 관리자와 의존성 관리 시스템과 통합되어 있지않음
• 데이터플로우 엔진은 조인 이외의 영역에서 선언적인 기능을 통합하는 것이 더 많은 장점이 있다고 판단함
  • 콜백 함수에 간단한 필터링 조건이 포함되어 있거나 레코드에서 일부 필드를 선택하는 경우,
    • 매 레코드마다 함수를 호출하는 것에 상당한 CPU 오버헤드가 발생
    • 이러한 간단한 필터링 및 매핑 작업을 선언적으로 표현하면, 쿼리 옵티마이저는 필요한 row만 디스크에서 읽을 수 있음

Specialization for different domains

• 임의의 코드를 실행할 수 있는 확장성은 유용하지만, 표준 처리 패턴의 재사용 가능한 구현도 필요함
• 머신 러닝 애플리케이션에 필요한 통계 및 수치 알고리즘에 대한 재사용 가능한 구현이 등장하고있음
  • Mahout은 MapReduce, Spark 및 Flink 위에서 머신 러닝을 위한 다양한 알고리즘을 구현
  • MADlib은 관계형 MPP 데이터베이스(Apache HAWQ) 내에서 유사한 기능을 구현
• 공간 알고리즘 및 유사성 검색과 같은 다양한 도메인에서 배치 처리 엔진이 사용됨
  • 배치 처리 시스템과 MPP 데이터베이스는 기능성과 선언적 연산자를 통해 서로 유사해지고 있음

[wookiist]

10. 일괄 처리

유닉스 도구로 일괄 처리

• 익혀둘만한 유닉스 도구 : awk, sed, grep , sort , uniq , xargs
• 이렇게 좋은 유닉스 도구에도,, stdin과 stdout을 사용할 때 제약사항이 있긴 함
  • 여러 개의 입력을 받거나, 여러 개의 출력이 필요할 때는 불가능하진 않지만,, 까다로움
• 유닉스 도구가 좋은 이유는 진행 사항 파악이 쉽기 때문임

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

• HDFS외에도 다양한 분산 파일 시스템이 있는데, 이번 장에서 배우는 예제는 대부분의 다른 분산 파일 시스템에도 적용됨
• HDFS 는 비공유 원칙을 기반으로 함. NAS와 SAN 아키텍처에서 사용하는 공유 디스크 방식과는 반대
• 리듀스 사이드 조인과 그룹화
  • 조인 - 레코드 간의 연관 관계를 처리하는 작어븡로, 외래 키, 문서 참조 또는 그래프 모델의 edge와 같은 연관 관계에서 필요
  • 조인은 연관 관계 양쪽에 있는 레코드에 접근해야 할 때 필요
    • MR 에서는 일반적인 의미의 인덱스가 없어서, 전체 테이블 스캔 방식을 사용함 (풀스캔)
  • 배치 맥락에서 조인은 모든 사용자의 데이터를 처리하는 것을 의미함
• 정렬 병합 조인
  • 매퍼 출력이 키로 정렬된 후에, 리듀서가 조인 양측의 정렬된 레코드 목록을 병합하는 조인임
• 쏠림
  • 같은 키를 가지는 모든 레코드를 같은 장소로 모으면 Hot Spot, Hot Key가 됨
  • 이를 피하기 위한 알고리즘으로 Hive는 핫 키를 테이블 메타데이터에 명시적으로 지정하고, 핫 키와 관련된 레코드를 나머지 키와는 별도 파일에 저장함. 해당 테이블에서 조인할 때, 핫 키를 가지는 레코드는 map-side join을 이용해 처리 함
• 맵 사이드 조인
  • 앞선 조인들은 reduce-side join임 (실제 조인 로직을 리듀서에서 수행하기 때문) → 입력 데이터에 대한 특정 가정이 필요 없고, 어떤 데이터든 매퍼는 데이터를 조인할 준비할 수가 있음.
    • 그러나 정렬 뒤, 리듀서로 복사한 뒤, 리듀서 입력을 병합하는 모든 과정의 비용이 상당함
  • 반면, 입력 데이터에 대해 특정 가정이 가능하면 map-side join을 수행할 수 있음. (리듀서, 정렬 작업 없음)
    • 각 매퍼가 분산 파일 시스템에서 단순히 입력 파일 블럭 하나를 읽어 다시 해당 분산 파일 시스템에 출력하는 것이 전부
• 브로드캐스트 해시 조인
  • map-side join은 작은 데이텃셋과 매우 큰 데이터셋을 조인할 때 적용할 수 있음!
    • 특히, 작은 데이터셋의 크기가, 그 전체를 각 mapper의 메모리에 적재가 가능할 정도라면!!!
    • 이를 모든 mapper가 시작할 때, in-memory 해시 테이블에 넣어버림.
    • 그러면 조인할 때, mapper는 작은 테이블의 내용을 쉽게 모두 스캔할 수 있고, 간단하게 조회할 수 있음
  • 브로드캐스트라는 뜻은, 큰 입력의 파티션 하나를 담당하는 각 매퍼는 작은 입력 전체를 읽는다는 것을 의미
• 파티션 해시 조인
  • 같은 방식으로, 맵 사이드 조인의 입력을 파티셔닝하면,, 해시 조인 접근법을 각 파티션에 독립적으로 적용할 수 있음
    • 예를 들어, 3번 매퍼가 ID가 3으로 끝나는 모든 사용자를 해시 테이블에 올리고, ID가 3으로 끝나는 각 사용자의 활동 이벤트 모두를 스캔하는 방식
  • Hive에선 버킷 맵 조인이라고 함
• 맵 사이드 병합 조인
  • 입력 데이터셋이 같은 방식으로 파티셔닝됐을 뿐 아니라, 같은 키를 기준으로 정렬됐다면, 변형된 맵 사이드 조인을 적용할 수 있음
    • 입력 크기가 메모리에 적재 가능한지 고려할 필요가 없음
      • 매퍼는 리듀서에서 일반적으로 수행하는 것과 동일한 병합 연산을 수행할 수 있기 때문

하둡과 분산 데이터베이스의 비교

• HDFS는 파일시스템이며 MR은 Unix Process의 구현체로 보면 됨
• MPP DB는 저장, query plan, 스케줄링, 쿼리 실행에 대한 통합 소프트웨어
  • MPP DB는 병렬 실행을 위해 설계되었고, 분석 SQl 쿼리에 중점을 두고 있음
  • 하둡은 ETL 프로세스에 자주 사용되며 데이터를 정제하고 변환하여 분석용 MPP DW에 갖고 옴
  • 작업중에 하나만 실패해도 전체 실패임
• MR을 사용하면 사용자가 직접 코드를 실행할 수 있으며, SQL 쿼리 외에도 다양한 형태로 배치 프로세스 작성이 가능
  • MR은 개별 작업에서의 장애가 괜찮음 (디스크에 데이터를 쓰니까)
• 하둡 생태계에는 HBase와 같은 OLTP DB와 Impala와 같은 분석용 DB가 있음

맵리듀스를 넘어

• 분산 파일 시스템 상에 있는 파일들은 중간 상태에 있는 경우가 있다.
  • 예를 들어, 추천 시스템을 구축할 때, 복잡한 워크플로는 MR 작업이 50-100개 정도로 구성되는데, 여기에는 많은 중간 상태가 존재한다.
• 이런 중간 상태를 파일로 기록하는 과정을 ‘구체화’라고 한다.
  • 그러나 유닉스 파이프에 비해 단점이 있다.
    • MR 작업은 입력을 생성하기 위한 모든 선행 작업이 완료되었을 때만 시작 가능. 반면, 유닉스 파이프로 연결된 프로세스는 동시에 시작되고, 생산되는 즉시 소비되어 출력됨
    • Mapper 작업은 종종 중복되기도 함
    • 분산 파일 시스템에서 중간 상태를 저장하는 것은, 중간 상태 파일이 복제되었음을 의미하므로, 과잉조치라고 볼 수 있음
• 이런 문제를 해결하기 위해, Spark, Tez, Flink 가 분산 일괄 처리 연산 분야에서 널리 쓰이고 있음
  • 전체 워크플로를 독립된 하위 작업으로 나누지 않고 작업 하나로서 다룬다는 점
• HDFS에 중간 상태를 쓰지 않는 엔진들은 내결함성 확보를 위해, 아직 유효한 데이터로부터 계산을 다시 해서 복구
  • 스파크는 RDD 추상화를 이용해 데이터 추상화
  • 플링크는 연산자 상태를 checkpoint를 남겨서 연산자 수행 재개
  • 다운스트림의 연산으로 장애가 전파되는 것을 피하려면, 연산자를 결정적으로 만드는 것이 좋음
• 일괄 처리 그래프를 최적화하기 위해 벌크 동기식 병렬이 널리 사용되며, Apache Giraph, Spark Graph X API, Flink Gelly API 등이 BSP 구현체며, 이는 프리글 (pregel) 모델이라고도 부름
  • 액터 모델과 살짝 비슷하며, MR과의 차이점은, 정점이 반복에서 사용한 메모리 상태를 기억하고 있다는 것. 그래서 정점은 새로 들어오는 메시지만 처리하면 됨

[dns02023]

일괄 처리

• 데이터 처리 시스템의 3가지 방식
  • 서비스(온라인 시스템)
  • 일괄 처리 시스템 (오프라인 시스템)
  • 스트림 처리 시스템 (준실시간 시스템)

유닉스 도구로 일괄처리하기

• 단순로그 분석
• 연쇄 명령 대 맞춤형 프로그램
• 정렬 대 인메모리 집계
• 유닉스 도구를 사용하면, 진행 사항을 파악하기가 상당히 쉽지만, 단일 장비에서만 실행된다는 제약이 존재하여, 하둡같은 도구가 필요함

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

• 맵리듀스는 수천 대의 장비로 분산 실행
• 분산 파일 시스템 상의 파일을 입력과 출력으로 사용
• 주로 HDFS를 다룰 예정
• HDFS는 비공유 원칙을 기반
• 장비가 죽거나 디스크가 실패하는 경우에 대비하기 위해 파일 블록은 여러 장비에 복제
• HDFS는 확장성이 뛰어나므로, 대규모 확장이 가능
• 매퍼, 리듀서 두가지 콜백 함수를 구현해야 함
• 병렬로 수행하는 코드를 직접 작성하지 않고도 여러장비에서 동시에 처리가 가능
• 매퍼 입력 파일의 복제본이 있는 장비에서 작업을 수행하면, 데이터 가까이에서 연산하므로, 네트워크 부하 감소 지역성 증가
• 코드는 작업이 할당된 장비에 아직 존재하지 않기 떄문에 맵리듀스 프레임워크가 적절한 장비로 코드 복사
• 리듀서를 기준으로 파티셔닝하고 정렬한 뒤 매퍼로부터 데이터 파티션을 복사하는 셔플 과정 발생
• 워크플로 도구 등을 통해 여러 맵리듀스 작업들을 연결하여 구성하는 방식이 일반적 (우지, 아즈카반, 루이지, 에어플로, 핀볼 등등)

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

정렬 병합 조인

• 매퍼 출력이 키로 정렬된 후에 리듀서가 조인의 양측의 정렬된 레코드 목록을 병합
• 필요한 데이터를 사전에 정렬하기 떄문에, 처리량 높게 유지 및 메모리 부담 줄임

그룹화

• 특정 키로 레코드를 그룹화
• 각 그룹 내에서 집계 연산

쏠림 다루기

• 키 하나에 너무 많은 데이터가 연관된 경우
• 리듀서 한개에 핫키에 대한 데이터들이 쏠리면 핫스팟 현상이라고 함
• 이 떄 쏠린 조인 메서드 등으로 핫스팟을 완화함

맵 사이드 조인

• 입력 데이터에 대해 특정 가정이 가능하다면 맵사이드 조인 기법으로 조인을 더 빠르게 수행 가능

  브로드캐스트 해시 조인

• 작은 데이터셋과 매우 큰 데이터셋을 조인하는 경우 적용
• 작은 데이터셋을 인메모리 해시 테이블에 넣음
• 큰 입력의 파티션 하나를 담당하는 각 매퍼는 작은 입력 전체를 읽는 것

파티션 해시 조인

• 조인할 두 입력 모두를 같은 키와 같은 해시 함수를 기반으로 같은 수로 파티셔닝해야 작동
• 제대로 파티셔닝 했다면, 조인할 레코드 모두가 같은 번호의 파티션에 위치
• 각 매퍼의 해시 테이블에 적재해야할 데이터 향을 줄일 수 있음

맵 사이드 병합 조인

• 파티셔닝 뿐만 아니라 같은 키를 기준으로 정렬되었다면 변형된 맵사이드 조인 적용 가능
• 오름차순으로 양쪽 입력 모두를 점진적으로 읽어 키가 동일한 레코드를 맞춘다.

하둡과 분산 데이터베이스의 비교

• 하둡은 유닉스의 분산 버전과 다소 비슷
• MPP 데이터베이스에서 이미 비슷한 방식으로 구현
• 맵리듀스와의 가장 큰 차이점은 장비 클러스터에서 분석 SQL질의를 병렬로 수행하는 것이 초점을 둔다.
• 데이터베이스는 특정모델을 따라 데이터를 구조화해야 하지만, 분산 파일시스템의 파일은 연속된 바이트일 뿐 어떤 형태던지 가능하다.
• 맵리듀스를 이용하면 엔지니어는 자신이 작성한 코드를 대용량 데이터셋 상에서 쉽게 실행할 수 있음
• 맵리듀스는 맵 또는 리듀스 태스크의 실패를 견딜 수 있다.

맵리듀스를 넘어

• 일괄 처리 방법의 대안을 살펴 봄

  중간 상태 구체화

• 워크플로 맵리듀스 작업이 많을 떄 많은 중간 상태가 존재
• 이를 파일로 기록하는 과정을 구체화라고 함

데이터플로 엔진

• 스파크, 테즈, 플링크 등가 같이 맵리듀스에 있는 문제들을 해결하기 위해 분산 일괄처리 엔진들이 등장했음
• 전체 워크플로를 독립된 하위작업으로 나누지 않고 작업 하나로서 다룬다는 점
• 데이터 흐름을 명시적으로 모델링하기 떄문에 데이터 플로 엔진이라고 부름
• 맵리듀스 워크플로와 동일한 연산을 데이터플로 엔진을 이용해 구현 가능하며, 최적화로 인해 수행속도가 훨씬 빠름

내결함성

• 중간 상태 구체화를 통해 내구성을 이점으로 가짐
• 스파크 플링크 테즈는 장비에 있던 중간 상태를 잃게 되면, 아직 유효한 데이터로부터 계산을 다시해서 복구함

구체화에 대한 논의

• 작업 완료 시 출력을 다른 사용자들이 사용할 수 있도록 지속성이 있는 어떤 곳으로 보내야 함
• 데이터 플로 엔진을 사용할 때 HDFS 상에 구체화된 데이터셋은 보통 작업의 입력과 최종 출력

그래프와 반복처리

• 그래프는 분산 파일 시스템에 정점과 간선 목록이 포함된 파일의 형태로 저장할 수 있음
• 맵리듀스는 그래프 저장에 적합하지 않음

프리글 처리 모델

• 일괄 처리 그래프를 최적화하는 방법으로 벌크 동기식 병렬 연산 모델이 널리 사용 됨
• 프리글 모델로 불리기도 함
• 스파크 그래프 X API, 플링크 젤리 API 등

고수준 API와 언어

• 테즈, 스파크, 플링크 등 자체 고수준 데이터플로 API를 가진 것들이 등장
• 코드를 적게 작성해도 되며, 대화식 사용도 지원하므로 효율적
• 선언형 질의언어로 전환
• 점차 광범위한 영역에서 필요한 알고리즘을 분산 수행하는 데 까지 사용