[고급 프로그래밍] Week 6: Hadoop의 MapReduce 구현과 Python을 이용한 예제

Day 1: Hadoop 소개

• 강의 내용:
  • Hadoop의 개념
    • Hadoop이란 무엇인가?
    • Hadoop의 주요 구성 요소
  • Hadoop 아키텍처
    • HDFS (Hadoop Distributed File System)
    • YARN (Yet Another Resource Negotiator)
    • Hadoop MapReduce
  • Hadoop의 장점과 단점
    • 대규모 데이터 처리의 장점
    • 복잡성과 비용
• 실습:
  • Hadoop 환경 설정 및 설치 안내

### Hadoop 설치 및 설정
1. Hadoop 다운로드 및 설치
   - https://hadoop.apache.org/ 에서 다운로드
   - 설치 및 환경 변수 설정
2. Hadoop 클러스터 설정
   - HDFS 및 YARN 구성 파일 설정
   - 네임노드 및 데이터노드 시작

 

Day 2: Hadoop의 MapReduce 구현

• 강의 내용:
  • Hadoop에서의 MapReduce
    • Map 단계와 Reduce 단계
    • 입력 형식과 출력 형식
  • Hadoop MapReduce 프로그래밍 모델
    • Mapper 클래스와 Reducer 클래스
    • 드라이버 클래스
  • Hadoop MapReduce의 작업 흐름
    • 입력 분할, 맵핑, 셔플링, 리듀싱
• 실습:
  • Hadoop MapReduce 예제 프로그램 작성

// WordCount.java
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

    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 {
            String[] tokens = value.toString().split("\\s+");
            for (String token : tokens) {
                word.set(token);
                context.write(word, one);
            }
        }
    }

    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);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

 

Day 3: Python을 이용한 MapReduce

• 강의 내용:
  • Hadoop Streaming
    • Hadoop Streaming이란 무엇인가?
    • Hadoop Streaming을 통한 Python 스크립트 사용
  • Python MapReduce 프로그램 작성
    • Mapper 스크립트와 Reducer 스크립트
  • Hadoop Streaming 작업 흐름
    • 입력, 맵핑, 셔플링, 리듀싱, 출력
• 실습:
  • Python을 사용한 Hadoop Streaming 예제

# mapper.py
import sys

for line in sys.stdin:
    for word in line.strip().split():
        print(f"{word}\t1")

# reducer.py
import sys
from collections import defaultdict

word_count = defaultdict(int)

for line in sys.stdin:
    word, count = line.strip().split('\t')
    word_count[word] += int(count)

for word, count in word_count.items():
    print(f"{word}\t{count}")

 

# Hadoop Streaming 실행 명령
hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-*.jar \
  -input input_dir \
  -output output_dir \
  -mapper mapper.py \
  -reducer reducer.py \
  -file mapper.py \
  -file reducer.py

 

Day 4: MapReduce 성능 최적화

• 강의 내용:
  • 성능 최적화 기법
    • 입력 데이터 분할 최적화
    • 맵퍼와 리듀서의 효율적 배치
  • 결과 병합 (Combiner)
    • Combiner의 역할과 사용 방법
  • Hadoop 설정 최적화
    • 메모리 설정, 블록 크기 조정
• 실습:
  • 성능 최적화를 적용한 MapReduce 예제

// WordCount.java (with Combiner)
import java.io.IOException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class WordCount {

    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 {
            String[] tokens = value.toString().split("\\s+");
            for (String token : tokens) {
                word.set(token);
                context.write(word, one);
            }
        }
    }

    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);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

 

Day 5: MapReduce 종합 연습

• 강의 내용:
  • 종합 연습 문제 풀이
    • Hadoop MapReduce와 Python MapReduce를 사용한 문제 해결
  • MapReduce의 응용
    • 실생활 예제에서의 MapReduce 활용
• 실습:
  • 종합 연습 문제 해결 및 결과 분석

### 종합 연습 문제 예시
1. Hadoop MapReduce를 사용하여 대규모 텍스트 데이터를 분석하세요.
2. Python Hadoop Streaming을 사용하여 로그 파일을 처리하세요.
3. MapReduce 작업의 성능을 최적화하여 대규모 데이터를 효율적으로 처리하세요.

 

Day 6: 프로젝트 준비

• 강의 내용:
  • 프로젝트 주제 선정 및 요구사항 분석
    • 프로젝트 주제 및 요구사항 확정
    • 프로젝트 설계 및 계획 수립
  • 프로젝트 구현 준비
    • 데이터 구조 및 알고리즘 설계
    • MapReduce 작업 관리 및 최적화 계획
• 실습:
  • 프로젝트 주제 및 요구사항 분석
  • 프로젝트 설계 및 계획 수립

### 프로젝트 주제 예시
1. Hadoop 기반 분산 데이터 처리 플랫폼 개발
2. Python MapReduce를 사용한 대규모 데이터 분석 도구 구축

### 프로젝트 요구사항 예시
1. Hadoop 기반 분산 데이터 처리 플랫폼:
   - 대규모 데이터를 분산 처리
   - MapReduce 작업을 통한 데이터 분석
   - 분석 결과 저장 및 시각화

2. Python MapReduce를 사용한 대규모 데이터 분석 도구:
   - 대규모 텍스트 데이터를 분할하여 처리
   - 데이터 필터링 및 변환
   - 분석 결과 저장 및 시각화

### 프로젝트 설계 및 계획 예시
1. 데이터 입력 모듈 구현
2. MapReduce 알고리즘 구현
3. 데이터 출력 및 성능 분석 모듈 구현

 

이 강의는 MapReduce 프로그래밍 모델, 특히 Hadoop의 MapReduce 구현과 Python을 이용한 MapReduce 예제의 기본 개념과 구현을 익히는 것을 목표로 하며, 각 강의는 이론과 실습을 포함합니다. 추가적인 주제나 다음 주차에 대한 상세 강의를 원하시면 말씀해 주세요!