[CS 스터디] Hadoop

Apache Hadoop

• High Availability Distributed Object Oriented Platform = 고가용성 분산형 객체 지향적 플랫폼
• 객체 지향적 작업을 병렬 분산하여 고가용성을 확보할 수 있는 플랫폼
• Java 기반 소프트웨어 오픈소스 플랫폼
• 빅데이터 애플리케이션용 데이터 처리와 스토리지를 관리하는 역할
• 구조적, 비구조적 데이터를 처리할 수 있으며, 단 한 대의 서버에서 시스템 수천 대 규모로 안정적으로 확장한다.
• HDFS(분산 파일 시스템)과 MapReduce를 구현한 것

어떻게 생기게 되었나?

끊임없이 늘어나는 빅데이터를 처리하고 웹 결과를 더 빨리 제공해야 한다는 필요성이 절실해지면서 탄생

고가의 서버들은 가격 관점에서 선형으로 성능이 증가하지 않기 때문에 아주 많은 값싼 서버들을 사용하는 것을 선호

영향?

기업에서 엄청난 양의 데이터를 저장, 처리할 수 있게 되었고, 컴퓨팅 파워를 증가하고 내결함성, 데이터 관리 유연성, DW에 비해 저렴한 비용은 물론 뛰어난 확장성까지 얻게 되었다.

주요 구성 요소

MapReduce

SW의 수행을 분산한다.

HDFS

데이터를 분산한다.

---

MapReduce

방대한 용량의 구조적, 비구조적 데이터를 처리할 때 애플리케이션을 쉽게 작성할 수 있는 프레임워크

값 싼 컴퓨터를 모아서 클러스터를 만들고 여기서 빅데이터 처리를 위한 Scalable한 병렬 SW의 구현을 쉽게 할 수 있도록 도와준다.

분산 프로그래밍의 복잡성을 없앤다.

메모리에 데이터를 거의 저장하지 않는다. 데이터를 읽고 HDFS에 기록해야 한다.

MapReduce 작업을 여러 번 할 때, 작업 간에 데이터를 전달할 수 없기 때문에 HDFS를 통한 데이터 공유가 필요하다. 이로 인해 처리 병목 현상이 발생한다.

 

1. Map 단계
   • 제일 먼저 수행되는 단계
   • 병렬 처리 작업 간에 데이터가 분할된다.
   • key-value 쌍 형식으로 결과를 출력하고 여러 머신에 나누어 보낸다. 이때, 같은 key를 가진 key-value 쌍은 같은 머신으로 보내진다.
   • 하나의 머신에서는 직렬로 이루어지지만 이런 동작이 여러 대의 머신에서 이루어지기 때문에 병렬이라고 할 수 있다.
   • 데이터 처리의 비즈니스 로직
2. Shuffling 단계
   • 모든 머신에서 Map 단계가 다 끝나면 수행되는 단계
   • Map 단계에서 받은 key-value 쌍을 key를 기준으로 정렬한다.
   • 같은 key를 가진 key-value 쌍을 모아서 value_list를 만든다.
   • key-value_list 쌍 형식으로 key에 따라 여러 머신에 분산해서 보낸다.
3. Reduce 단계
   • 모든 머신에서 shuffling 단계가 다 끝나면 수행되는 단계
   • 각각의 key-value_list 쌍 마다 reduce 함수가 호출된다.
   • 하나의 reduce 함수가 끝나면 다음 key-value_list 쌍에 대해 reduce 함수가 호출된다.
   • Map 함수의 중간 데이터 출력을 요약하여 최종 결과를 산출한다.

"WordCount" MapReduce 코드 

1. Mapper

/*
    Object, Text : input key-value pair type (always same (to get a line of input file))
    Text, IntWritable : output key-value pair type
    */
    public static class TokenizerMapper
            extends Mapper<Object,Text,Text,IntWritable> {
        // variable declairations
        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());
                // emit a key-value pair
                context.write(word,one);
            }
        }
    }

2. Reducer

    /*
    Text, IntWritable : input key type and the value type of input value list
    Text, IntWritable : output key-value pair type
    */
    public static class IntSumReducer
            extends Reducer<Text,IntWritable,Text,IntWritable> {
        // variables
        private IntWritable result = new IntWritable();
        // key : a disticnt word
        // values :  Iterable type (data list)
        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);
        }
    }

3. Main

    /* Main function */
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();  // job을 수행하기 위한 설정 초기화
        String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
        if ( otherArgs.length != 2 ) {
            System.err.println("Usage: <in> <out>");
            System.exit(2);
        }
        Job job = new Job(conf,"word count");  // 따옴표 안은 설명
        job.setJarByClass(Wordcount.class);  // job을 수행할 class 선언, 파일명.class
        // let hadoop know my map and reduce classes
        job.setMapperClass(TokenizerMapper.class);  // Map class 선언, 위에서 작성한 class 명
        job.setReducerClass(IntSumReducer.class);  // Reduce class 선언

				// combine 함수를 쓰고 싶으면 아래와 같이 하면 된다.
				// job.setCombinerClass(IntSumReducer.class); // Combiner class 선언

        job.setOutputKeyClass(Text.class);  // reduce의 Output key type 선언
        job.setOutputValueClass(IntWritable.class);  // reduce의 Output value type 선언
				// Map과 reduce의 output이 같을 경우에 Map output에 대한 지정을 생략할 수 있다.
				// job.setMapOutputKeyClass(Text.class);
				// job.setMapOutputValueClass(IntWritable.class);

        // set number of reduces
        job.setNumReduceTasks(2);  // 동시에 수행되는 reducer의 개수
        // set input and output directories
        FileInputFormat.addInputPath(job,new Path(otherArgs[0]));  // 입력데이터가 있는 path
        FileOutputFormat.setOutputPath(job,new Path(otherArgs[1]));  // 결과를 출력할 path
        System.exit(job.waitForCompletion(true) ? 0 : 1 );  // 실행
    }

*Combine 함수

reduce 함수와 유사한 함수

Map 단계에서 Map 함수의 출력 크기를 줄여 Shuffling 단계와 Reduce 단계의 비용을 줄여주는데 사용된다.

---

HDFS

• Hadoop Distributed File System
• hadoop 프레임워크를 위해 자바 언어로 작성된 HDFS
• 빅데이터를 수천 대의 값싼 컴퓨터에서 병렬 처리하기 위해 분산해서 저장하는 곳
• 여러 기계에 대용량 파일들을 나눠서 저장한다.
• fault tolerance를 위해 데이터를 여러 서버에 중복해서 저장함으로써 데이터 안정성을 얻는다.

* fault tolerance 

시스템을 구성하는 일부에서 fault(결함)이나 failure(고장)가 발생해도 정상적이거나 부분적으로 기능을 수행할 수 있는 것

 

master/slave 구조

NameNode

• Hadoop HDFS의 master 데몬
• 파일 시스템을 관리하고 클라이언트의 접근을 제어한다.
• 파일 시스템 네임 스페이스를 관리하고 이 파일들에 대한 클라이언트 접근을 제어한다.
• 파일과 디렉터리의 open, close, rename 등 파일 시스템의 namespace의 여러 기능을 수행한다.
• DataNode와 block의 mapping을 결정한다.
• block번호와 위치 등과 같은 메타데이터를 저장한다.

DataNode

• Hadoop HDFS의 slave 데몬
• 컴퓨터에 들어있는 데이터에 접근할 수 있게 한다.
• 클라이언트가 요청한 read 및 write 함수를 실행하며 실제 데이터나 block을저장한다.
• 하나의 파일은 하나 이상의 block으로 나뉘어 있고 이 block들은 DataNode에 저장되어 있다.
• Master NameNode의 명령에 따라 복제본 생성, 삭제 및 복제를 담당한다.
• 두 개의 시스템 파일로 구성된다. 하나는 데이터에 사용하고 나머지 하나는 블록 메타데이터를 기록하는데 사용한다.

 

애플리케이션이 시작되면 Master와 Slave 데몬 사이에 handshake를 하여 Namespace와 Software version을 확인한다.

 

Hadoop의 장점

1. 확장성
   • 분산형 환경에서 동작하기 때문에 확장 가능
2. 복원력
   • HDFS는 의도적으로 중복 저장하기 때문에 hw나 sw 오류 가능성에 대비
3. 유연성
   • 형식에 구애받지 않고 데이터를 저장할 수 있음

Hadoop의 단점

1. 복잡함
   • hadoop 아키텍처는 상당한 수준의 전문 지식과 리소스가 있어야 설치하고 유지할 수 있다.
2. 성능
   • disk로부터 read, write를 자주 사용하기 때문에 Apache Spark에 비해 시간이 오래 걸리고 비효율적

---

 

Apache Spark

• 빅데이터 작업에 사용되는 오픈 소스 분석 엔진
• 암시적 데이터 병렬성과 장애 허용과 더불어 완전한 클러스터를 프로그래밍하기 위한 인터페이스
• 배치와 실시간 분석 및 데이터 처리 작업을 모두 처리할 수 있다.
• 인메모리 엔진을 사용해 처리 속도가 높다.
• MapReduce 모델을 확장해 대화형 쿼리 및 스트림 처리를 포함하는 더 많은 유형에서 효율적으로 사용 가능
• 머신러닝, 스트림 처리, 그래프 처리를 위한 여러 라이브러리

어떻게 생기게 되었나?

Hadoop 시스템에서 작업 처리 속도를 높이는 방법을 찾다가

기존에 MapReduce를 진행할 때 파일 시스템을 이용하였는데 이는 데이터 공유가 많이 느리다.

(hadoop 시스템의 대부분은 HDFS 읽기, 쓰기 작업을 수행하는데 90% 이상을 사용하고 있다고 한다.)

주요 구성요소

Spark Core와 일련의 라이브러리

---

Spark Core

• Apache Spark의 핵심.
• 분산 작업 전송, 예약 및 I/O 기능을 제공
• RDD 개념을 기본 데이터 유형으로 사용

RDD

• Resilient Distributed Dataset
• 분산 변경 불가능한 객체 모음
• 메모리 내 처리 연산을 지원한다.
• 사용자로부터 계산 복잡성의 대부분을 숨길 수 있음
• Spark으 모든 작업은 새로운 RDD를 만들거나 존재하는 RDD를 변경하거나 결과 계산을 위해 RDD를 연산하는 것
• https://artist-developer.tistory.com/17

Spark의 장점

1. Speed
   • 여러 병렬 작업에서 데이터를 메모리에 캐싱하여 매우 빠르게 실행된다.  디스크에서의 읽기,쓰기 작업 횟수를 줄여서 매우빠름
2. Ease of Use
   • 여러 프로그래밍 언어를 지원한다. 
   • Java, Scala, Python, R
3. Advanced Analytics
   • SQL 쿼리, 머신 러닝, 스트림 처리 및 그래프 처리를 지원한다.

---