[Hadoop] Streaming API

Streaming API를 이번 포스팅에서는 알아보자. Streaming API는 왜 쓰는가? 조금 더 편하게 MapReduce를 수행하려고 쓰는 것이다. Streaming API를 이용하면 일일이 MapReduce를 작성할 없이 간단하게 작성할 수 있다. (물론 프로그래밍이 약간 들어가긴 한다. 그래도 상대적으로 무지하게 간단하게 끝낼 수 있다.) 가장 기본적인 Streaming 이용은 아래와 같다.

간단하게 Unix 명령어를 이용한 Streaming이다.

./bin/hadoop jar ./share/hadoop/tools/lib/hadoop-streaming-2.5.2.jar

-input /example/MyJob/input/cite75_99.txt 

-output output 

-mapper 'cut -f 2 -d ,' 

-reducer 'uniq'

위의 명령어를 조금 설명해보자. 우선 이용하려는 jar 파일은 기본적으로 Hadoop package에 예제로 주어지는 jar 파일이다. input file의 형태는 

"CITING","CITED"

3858241,956203

3858241,1324234

3858241,3398406

3858241,3557384

....

위와 같다. CITING이라는 Key값과 CITED라는 Value값을 가지고 있다. output은 HDFS에서의 경로이다. 중요한 부분은 mapper 그리고 reducer 부분이다.

mapper Function

cut -f 2 -d,

무슨 뜻일까? input 문자열들에 대해서 comma(,) character를 구분자로 하여 구분된 문자열의 두 번째 Field를 사용하겠다는 것이다. 그러면 CITED 부분만 뽑혀 나오겠다.

"CITED"

956203

1324234

3398406

3557384

reducer Function

이 결과물을 Reducer의 input으로 보내어 uniq를 실행한다. 중복되는 것을 지우는 것이다. 따라서 CITED의 고유의 값이 이 MapReduce의 결과물이 된다.

Streaming을 이용하면 무지하게 편안하게 MapReduce를 할 수 있다. 비록 Hadoop을 이용한 실전 업무를 해본 경험은 없지만, 책에서 배우기로는(ㅋㅋㅋ) Hadoop의 Streaming은 자주 쓰인다고 한다. 왜? 간편하니까!

이번에는 Python을 이용하여 Streaming을 이해해보자.

./bin/hadoop jar ./share/hadoop/tools/lib/hadoop-streaming-2.5.2.jar -D mapred.reduce.tasks=1 -input /example/MyJob/input/apat63_99.txt -output /example/MyJob/output -mapper 'TopAttribute.py 8 4' -reducer 'TopAttribute.py 0 4' -file /anything/programming/python/hadoop/TopAttribute.py

위의 Python Code를 위의 명령어로 실행하면 아래와 같은 결과물이 나온다.

참고로 apat63_99.txt는 아래와 같은 Data이다.

"PATENT","GYEAR","GDATE","APPYEAR","COUNTRY","POSTATE","ASSIGNEE","ASSCODE","CLAIMS","NCLASS","CAT","SUBCAT","CMADE","CRECEIVE","RATIOCIT","GENERAL","ORIGINAL","FWDAPLAG","BCKGTLAG","SELFCTUB","SELFCTLB","SECDUPBD","SECDLWBD"

3070801,1963,1096,,"BE","",,1,,269,6,69,,1,,0,,,,,,,

3070802,1963,1096,,"US","TX",,1,,2,6,63,,0,,,,,,,,,

3070803,1963,1096,,"US","IL",,1,,2,6,63,,9,,0.3704,,,,,,,

3070804,1963,1096,,"US","OH",,1,,2,6,63,,3,,0.6667,,,,,,,

3070805,1963,1096,,"US","CA",,1,,2,6,63,,1,,0,,,,,,,

3070806,1963,1096,,"US","PA",,1,,2,6,63,,0,,,,,,,,,

3070807,1963,1096,,"US","OH",,1,,623,3,39,,3,,0.4444,,,,,,,

3070808,1963,1096,,"US","IA",,1,,623,3,39,,4,,0.375,,,,,,,

3070809,1963,1096,,"US","AZ",,1,,4,6,65,,0,,,,,,,,,

.....

Field가 조금 많아서 복잡하겠지만 잘 보길 바란다.

MapReduce를 상대적으로 무지하게 쉽게 실행시킬 수 있다는 것에 동감하는가? 아직까지 내가 해본 MapReduce여봤자 몇 가지 안 되지만, 내가 느끼기로는 확실히 편하다. 개인적으로 느낀 것 중에 하나는, 그 많고 다양한 Hadoop의 Java Class 사용하지 않아도 된다는 것이 편했다. 더 정확히 말하면, input과 output의 Type 결정하는 부분에 있어서 Streaming으로 했을 경우 간단하게 할 수 있다. 

그런데 여기서 잠깐, Mapper라면은 기본적으로 Key/Value Pair를 적절하게 활용해야되지 않나? 그런데 위의 예제에서는 굳이 어떤 것이 Key이니 Value이니 가리지 않고 처리했다. Streaming에서도 Key와 Value를 나누기는 한다. 아래와 같이 Hadoop 공식 홈페이지에서 나와있다. 

By default~ 부분을 읽어보라. 기본적으로 '\t' character로 구분을 한다. 그런데 각 라인에 tab이 없을 경우 줄 내용 전체가 key가 되고 Value는 null이다. python script에서 어떤식으로 Key와 Value를 구분하여 사용하는지는 정확하게 모르겠다. 예를 들어 이전의 포스팅에서 Java example을 보면, mapper Function의 argument를 통해 Key/Value에 접근할 수 있다. 그런데 python에서는? 잘 모르겠다. (이 부분에 대해서 알고 계시는 분은 뎃글 남겨주시면 감사하겠습니다.)

한 가지 확실한 것은 python에서 MapReduce 작성이 훨씬 수월하게 느껴졌다는 점이다. 이번 포스팅은 여기서 마칠 생각이다. 주제는 Streaming을 이용하면 훨씬 수월하게 MapReduce를 작성할 수 있다는 점. 다음 포스팅에서는 Combiner에 대해서 이야기할 생각이다. Combiner는 MapReduce의 효율을 향상시킬 수 있는 한 가지 방법이다. 이 정도로 간단하게만 알고, 다음 포스팅을 기약하자.

Reference:

http://hadoop.apache.org/docs/r1.2.1/streaming.html