하둡 맵리듀스 Join 활용 예제

하둡 맵리듀스를 활용하다 보면 서로 다른 유형의 데이터 셋을 조인해야 하는 경우가 종종 있다. 

SQL에서 테이블간 조인을 생각해 보면 된다. 

Word Count 예제를 기반으로 맵리듀스의 조인을 고려해 보자. 

특정 단어의 개수를 세는데 파일 하나는 전체 기간을 대상으로 하고, 다른 파일은 월별로 각 단어의 개수를 나타낸다고 해보자. 

아래의 두 파일을 하나로 합쳐서 <날짜 단어 월별개수 전체개수> 형태로 합쳐서 출력하는 부분을 하둡 맵리듀스로 구현해보는 것이다. 

join1_FileA.txt

able,991
about,11
burger,15
actor,22

join1_FileB.txt

Jan-01 able,5
Feb-02 about,3
Mar-03 about,8
Apr-04 able,13
Feb-22 actor,3
Feb-23 burger,5
Mar-08 burger,2
Dec-15 able,100

맵 함수를 작성할 때 먼저 고려해야 할 사항은 어떤 것을 키(Key)로 설정할 것인가이다. 

위의 두 파일을 보면 조인 할 수 있는 공통 부분이 단어이므로 단어를 키로 설정하면 된다. 

그럼 파이썬으로 만든 맵 프로그램 소스를 살펴보자. 

join1_mapper.py

#!/usr/bin/env python import sys for line in sys.stdin: line = line.strip() key_value = line.split(",") key_in = key_value[0].split(" ") value_in = key_value[1] if len(key_in)>=2: date = key_in[0] word = key_in[1] value_out = date+" "+value_in print( '%s\t%s' % (word, value_out) ) else: print( '%s\t%s' % (key_in[0], value_in) )

그리고 리듀스 함수에서 합칠 때 값(Value)에 따라 적절하게 융합하면 된다. 

리듀스 함수의 소스는 다음과 같다.

join1_reducer.py 

#!/usr/bin/env python import sys prev_word = " " months = ['Jan','Feb','Mar','Apr','Jun','Jul','Aug','Sep','Nov','Dec'] dates_to_output = [] day_cnts_to_output = [] line_cnt = 0 for line in sys.stdin: line = line.strip() key_value = line.split('\t') line_cnt = line_cnt+1 curr_word = key_value[0] value_in = key_value[1] if curr_word != prev_word: if line_cnt>1: for i in range(len(dates_to_output)): print('{0} {1} {2} {3}'.format(dates_to_output[i],prev_word,day_cnts_to_output[i],curr_word_total_cnt)) dates_to_output =[] day_cnts_to_output=[] prev_word =curr_word if (value_in[0:3] in months): date_day =value_in.split() dates_to_output.append(date_day[0]) day_cnts_to_output.append(date_day[1]) else: curr_word_total_cnt = value_in for i in range(len(dates_to_output)): print('{0} {1} {2} {3}'.format(dates_to_output[i],prev_word,day_cnts_to_output[i],curr_word_total_cnt))

파이썬으로 생성한 맵리듀스 프로그램을 한번 실행해 보자. 

1. 새롭게 생성한 파이썬 소스를 실행가능하도록 설정한다. 

> chmod +x join1_mapper.py
> chmod +x join1_reducer.py
> ls -al join1_*.py

2. 하둡 파일 시스템의 /user/cloudera/input 디렉토리에 있는 기존 파일을 제거한다. 

> hdfs dfs -ls /user/cloudera/input
> hdfs dfs -rm /user/cloudera/input/*
> hdfs dfs -ls /user/cloudera/input

3. 맵리듀스 조인에 활용할 예제 파일을 하둡 파일 시스템으로 올린다.

> hdfs dfs -put join1_File*.txt /user/cloudera/input/
> hdfs dfs -ls /user/cloudera/input

4. 하둡 스트리밍을 활용하여 실행한다. 

> hadoop jar /usr/lib/hadoop-mapreduce/hadoop-streaming.jar \
  -input /user/cloudera/input \
  -output /user/cloudera/output_join \
  -mapper /home/cloudera/join1_mapper.py \
  -reducer /home/cloudera/join1_reducer.py

5. 실행 결과를 확인해 보자. 

> hdfs dfs -ls /user/cloudera/output_join
> hdfs dfs -cat /user/cloudera/output_join/part-00000

6. 맵리듀스의 결과와 파이프라인으로 실행한 결과를 비교해보면 순서는 다르지만 동일한 결과가 나오는 것을 확인할 수 있을 것이다. 

> cat join1_File*.txt | ./join1_mapper.py | sort | ./join1_reducer.py

위 예제는 University of California, San Diego의 Super Computer Center, Paul Rodriguez님의 강의를 참고했다.