📒 Hadoop (1)

📌 Hadoop

• 상용 HW로 구축된 컴퓨터 클러스터에서 매우 거대한 데이터셋의 분산 스토리지와 분산 처리를 위한 오픈소스 SW 플랫폼이다.

• An open source software platform for distributed storage and distributed processing of very large data sets on computer clusters built from commodity hardware

  • HDFS : 분산 파일 시스템
  • MapReduce: 분산 컴퓨팅 시스템

• 다수의 노드로 구성된 클러스터 시스템(Cluster)

  • 마치 하나의 거대한 컴퓨터처럼 동작
  • 사실은 다수의 컴퓨터들이 복잡한 SW로 통제된다.

• Hadoop 1.0

  • HDFS위에 MapReduce라는 분산 컴퓨팅 시스템이 도는 구조이다.
    -> MapReduce위에서 다양한 컴퓨팅 언어들이 만들어진다.

• Hadoop 2.0

  • 아키텍처가 크게 변경되었다.
    • YARN이란 이름의 분산처리 시스템위에서 동작하는 APP이 되었다.
    • Spark은 YARN위에서 APP LAYER로 실행되었다.

---

📌 HDFS - 분산 파일 시스템

• 데이터를 블록단위로 나누어 저장한다.
  -> 블록의 크기는 128MB(Default)

• 블록 복제 방식(Replication)

  • 각 블록은 3 군데에 중복 저장된다.
  • Fault tolerance를 보장할 수 있는 방식으로 이 블록들은 저장된다.

• Hadoop 2.0 NameNode 이중화지원

  • Active & Standby
    -> 둘 사이에 share edit log가 존재한다.
  • Secondary NameNode는 여전히 존재한다.

---

📌 MapReduce - 분산 컴퓨팅

• Hadoop 1.0

• 하나의 Job Tracker와 다수의 Task Tracker로 구성되었다.

  • Job Tracker가 일을 나누어서 다수의 Task Tracker에게 분배한다.
  • Task Tracker에서 병렬처리한다.

• MapReduce만 지원한다.

  • General한 시스템이 아니다.

---

📌 YARN

1. 분산 컴퓨팅 시스템 : Hadoop 2.0(YARN 1.0)

• 세부 리소스 관리가 가능한 범용 컴퓨팅 플레임 워크이다.

  • 리소스 매니저
    -> Job Scheduler, Application Manager

  • Node Manager

  • Container

    • App Master
    • Task

• Spark가 이 위에서 구현된다.

2. YARN의 동작

• 실행하려는 코드와 환경 정보를 RM(Resource Manager)에게 넘긴다.

  • 실행에 필요한 파일들은 Application ID에 해당하는 HDFS 폴더에 미리 복사된다.

• RM은 NM(Node Manager)으로부터 Container를 받아 AM(Application Master)을 실행한다.

  • AM은 프로그램 마다 하나씩 할당되는 프로그램 마스터에 해당한다.

• AM은 입력 데이터 처리에 필요한 리소스를 RM에게 요구한다.

  • RM은 Data Locality를 고려해서 리소스(Container)를 할당한다.

• AM은 할당받은 리소스를 NM을 통해 Container로 Launch하고 그 안에서 코드를 실행한다.

  • 이 때 실행에 필요한 파일들이 HDFS에서 Container가 있는 서버로 먼저 복사한다.

• 각 Task는 상황을 주기적으로 AM에게 보고한다. (Heartbeat)

  • Task가 실패하거나 보고가 오랜 시간 없으면 Task를 다른 Container로 재실행한다.

3. Hadoop 1.0 vs. Hadoop 2.0

• 하둡 2.0에서 소개된 클러스터 자원 관리자를 YARN이라고 부른다.
  

4. Hadoop 3.0의 특징

• YARN 2.0을 사용

  • YARN 프로그램들의 논리적인 그룹(플로우라고 부른다)으로 나눠서 자원관리가 가능하다.
  • 이를 통해 데이터 수집 프로세스와 데이터 서빙 프로세스를 나눠서 관리가능하다.
  • 타임라인 서버에서 HBase를 기본 스토리지로 사용한다.(Hadoop 2.1)

• 파일 시스템

  • NameNode의 경우 다수의 스탠바이 NameNode를 지원한다.

  • HDFS, S3, Azure Storage 이에외데 Azure Data Lake Storage등을 지원한다.

---

📌 MapReduce Programing

1. MapReduce Programing의 특징

• 데이터 셋은 Key, Value의 집합이며 변경 불가하다. (Immutable)

• 데이터조작은 map과 reduce 두 개의 오퍼레이션으로만 가능하다.

  • 이 두 오퍼레이션은 항상 하나의 쌍으로 연속으로 실행된다.
  • 이 두 오퍼레이션의 코드를 개발자가 채워주어야 한다.

• MapReduce System이 Map의 결과를 Reduce단으로 모아준다.

  • 이 단계를 보통 셔플링이라고 하며 네트워크단을 통한 데이터 이동이 생긴다.

2. Map과 Reduce

• Map: (k, v) -> [(k', v')*]
  • 입력은 시스템에 의해 주어지며 입력으로 지정된 HDFS 파일에서 넘어온다.
  • 키, 밸류 페어를 새로운 키, 밸류 페어 리스트로 변환(Transformation)
  • 츌력: 입력과 동일한 키, 밸류 페어를 그대로 출력해도 되고 출력이 없어도 된다.
• Reduce: (k', [v1', v2', v3', v4', ...]) -> (k'', v'')
  • 입력은 시스템에 의해 주어진다.
    -> 맵의 출력 중 같은 키를 갖는 키/밸류 페어를 시스템이 묶어서 입력으로 넣어준다.
  • 키와 밸류 리스트를 새로운 키/밸류 페어로 변환한다.
  • SQL의 Group by와 흡사하다.
  • 출력이 HDFS에 저장된다.

3. MapReduce Program 동작 예시

• Word Count
  

4. MapReduce Programing 예제

• Word Count Mapper

• Map: (k, v) -> [(k', v')*]

  • Transformation
  • 킬, 밸류 페어를 새로운 키, 밸류 페어 리스트로 변환

• Input: (100, "the brave yellow lion")

• Output: [("the", 1), ("brave", 1), ("yellow", 1), ("lion", 1)]

public static class TokenizerMapper 
     extends Mapper<Object, Text, Text, IntWritable>{
    
  private final static IntWritable one = new IntWritable(1);
  private Text word = new Text();
      
  public void map(Object key, Text value, Context context
                  ) throws IOException, InterruptedException {
    StringTokenizer itr = new StringTokenizer(value.toString());
    while (itr.hasMoreTokens()) {
      word.set(itr.nextToken());
      context.write(word, one);
    }
  }
}

5. MapReduce Programing 예제

• Word Count Reducer

• Reduce: (k', [v1', v2', v3', v4', ...]) -> (k'', v'')

  • SQL의 Group by와 동일하다
  • 키, 밸류 리스트를 새로운 키, 밸류 페어로 변환한다.

• Input: ("lion": [1, 1, 1])

• Output: ("lion": 3)

public static class IntSumReducer 
     extends Reducer<Text,IntWritable,Text,IntWritable> {
  private IntWritable result = new IntWritable();

  public void reduce(Text key, Iterable<IntWritable> values, 
                     Context context
                     ) throws IOException, InterruptedException {
    int sum = 0;
    for (IntWritable val : values) {
      sum += val.get();
    }
    result.set(sum);
    context.write(key, result);
  }
}

6. MapReduce: Shuffling and Sorting

• Shuffling

  • Mapper의 출력을 Reducer로 보내주는 프로세스를말한다.
  • 전송되는 데이터의 크기가 크면 네트워크 병목을 초래하고 시간이 오래걸린다.

• Sorting

  • 모든 Mapper의 출력을 Reducer가 받으면 이를 키별로 소팅한다.

7. MapReduce: Data Skew

• 병렬처리의 큰 의미가 없다.
• 가장 느린 Task가 전체 처리 속도를 결정한다.
• 특히 Reducer로 오는 데이터 크기는 큰 차이가 있을 수 있다.
  • Group By나 Join등이 이에 해당한다.
  • 처리 방식에 따라 Reducer의 수에 따라 메모리 에러등이 날 수 있다.
• 데이터 엔지니어가 고생하는 이유 중의 하나
  • 빅데이터 시스템에는 이 문제가 모두 존재한다.

8. MapReduce 프로그래밍의 문제점

• 낮은 생산성

  • 프로그래밍 모델이 가진 융통성 부족 (2가지 오퍼레이션만 지원한다.)
  • 튜닝/최적화가 쉽지 않다.
  • 예) 데이터 분포가 균등하지 않은 경우

• 배치작업 중심

  • 기본적으로 Low Latency가 아니라 Throughput에 초점이 맞춰진다.

9. MapReduce 대안들의 등장

• 더 범용적인 대용량 데이터 처리 프레임워크들의 등장
  • YARN, Spark
• SQL의 컴백: Hive, Presto등이 등장
  • Hive
    • MapReduce위에서 구현된다.
    • Throughput에 초점을 둔다.
    • 대용량 ETL에 적합하다.
  • Presto
    • Low Latency에서 초점을 둔다.
    • 메모리를 주로 사용한다.
    • Adhoc 쿼리에 적합하다.
    • AWS Athena가 Presto 기반이다.