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

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

으아. 무려 한시간 동안 쓴 거를 순간 날려먹었다. 다시 쓰자.

Hadoop의 Basic Example을 소개하려고 한다. 당연히 MapReduce를 사용하는 예제를 보여줄 것이고 그것에 대한 가장 기본적인 예제가 WordCount 이다. C에서의 HelloWorld라고 생각하면 된다.

Hadoop의 MapReduce를 수행하기 위해서는 당연히 Mapper Function과 Reducer Function이 필요하다. 그리고 그것을 수행시켜줄 main Function도 당연히 필요하다. 아래는 WordCount Example 예제이다. (Java로 쓰여져있다.)

보다 시피 map function과 reduce function이 있다. 그리고 뭐라고 쓰여있는 것인지는 모르겠지만 main function도 있다. 우리는 우선 main function 부터 알아볼 생각이다. main function에서 하는 일은 크게 아래와 같다.

1. JobConf를 선언한다.

2. JobConf를 통해서 Mapper를 설정한다.

3. JobConf를 통해서 Reducer를 설정한다.

4. JobConf를 통해서 Mapper와 Reducer의 input 그리고 output Type을 설정한다.

5. input과 output의 path 정보를 설정한다.  (argument 관련된 정보이다.)

6. JobClient object를 통해서 JobConf object를 실행시킨다. 이 때 사용하는 함수가 runJob이다. 

자꾸 JobConf JobConf하는데 JobConf가 무엇이냐? 여행사와 같은 에이전트 역할을 한다고 보면 된다. 우리가 해외여행을 갈 때는 보통 여행사에 Contact을 해서 숙소,일정 등을 잡고 여행사가 현지의 숙소, 비행일정등을 잡아준다. 마찬가지로 우리는 JobConf를 통해서 Hadoop의 MapReduce를 이용할 것이다. 우리는 JobConf에게 위의 1번부터 6번까지의 일을 해줄 것이며 나머지는 JobConf가 알아서 map Function 그리고 reducer Function을 Call 할 것이다. 

참고로 더욱 정확한 JobConf의 내용은 아래를 참고하자.

Hadoop 공식 사이트에서 Capture한 것이다.

굳이 위의 1번, 2번,3번, 6번은 설명하지 않겠다. 소스코드를 보면 금방 알 수 있다.

4. JobConf를 통해서 Mapper와 Reducer의 input 그리고 output Type을 설정한다.

Mapper와 Reducer는 당연히 각각의 input Type과 output Type이 있다. 그리고 당연한 이야기 이지만 Mapper의 output이 Reducer의 input으로 가기 때문에 그 두개의 Type은 같아야 한다. 예를 들어, Mapper의 output의 Key/Value output Type이 String/Int이면 Reducer의 Input의 Key/Value input Type 역시 String/Int 이어야 한다. 

위의 소스코드를 보면 Text.class, IntWritable.class를 볼 수 있을 것이다. 그 두 개가 각각 String과 Int 형에 대한 Hadoop의 object이다. Hadoop은 String과 Int를 자기 나름데로 Class로 구현하여 사용한다. 

JobConf를 통해서 input 그리고 output의 Type을 결정하는 부분은 아래와 같다.

conf.setInputFormat(TextInputFormat.class);

conf.setOutputFormat(TextOutputFormat.class);

conf.setOutputKeyClass(Text.class);

conf.setOutputValueClass(IntWritable.class);

TextInputFormat.class에 대해서 또 이야기 하고 넘어가야 하는 것이 있다. 아래의 표를 보자.

위의 Table이 모두 설명해주고 있으니 굳이 따로 추가적인 설명은 생략하겠다. 표에서 밑에 두 개의 Class는 적어도 한 번 써본 후 설명하려고 한다. 일단은 공백으로 놔두려고 한다.

5. input과 output의 path 정보를 설정한다.  (argument 관련된 정보이다.)

FileInputFormat.setInputPaths(conf, new Path(args[0]));

FileOutputFormat.setOutputPath(conf, new Path(args[1])); 

이 부분은 input과 output의 argument 위치를 말해주는 부분이다.

위의 그림은 Hadoop 공식 홈페이지에서 캡쳐한 것이니 참고하시길 바란다.

개괄적인 것은 모두 설명했으니 이제 WordCount를 이해해보자.

이 예제는 TextInputFormat.class를 사용하고 있기 때문에 Input의 Key/Value pair는 각각 LongWritable/Text Type을 가지게 된다. WordCount는 말 그대로 단어 세는 프로그램이다. 예를 들어 다음의 txt 파일이 Input으로 들어왔다고 해보자.

========= sample.txt =========

I love you

you do not love me I am sad

=============================

위의 txt 파일은 두 line이 있다. 한 줄 한 줄이 각각 Input이 될 것이다. I love you라는 하나의 Input과 you do not love me I am sad라는 하나의 Input이다. I love you는 TextInputFormat.class에서는 LongWritable Type의 Key와 Text Type의 Value로 구분된다. 이 때 LongWritable은 byte offset이므로 <10, "I love you"> 이렇게 된다. 두 번째 line은 <27, "you do not love me I am sad"> 이렇게 된다.

이 Input으로 map Function을 돌린다. 각각의 value를 스페이스로 구분하여 key로 놓고 그 값을 모두 one이라는 IntWritable object로 놓는다. 그러면 map Function의 결과값은 아래와 같게 된다.

첫번째 line :
<"I", 1>, <"love", 1>, <"you", 1>

두번째 line :
<"you", 1>, <"do", 1>, <"not", 1>, <"love", 1>, <"me", 1>, <"I", 1>, <"am", 1>, <"sad", 1>

이 때 하나만 집고 넘어가자. map Function의 Input 값에서 Key에 해당하는 정보는 map Function에서 전혀 쓰이지 않았다는 점이다. MapReduce 함수를 작성하다보면 이런 경우를 굉장히 자주 볼 수 있다. 이것이 옳은 것인지 그른 것인지는 잘 모르겠으나, 내가 몇 개 안되는 MapReduce를 작성해보고 느낀 바로는, 굳이 그렇게 나쁠 것까지는 없다? 정도의 의견을 조심스럽게 말하련다.

아무튼 위와 같은 형태로 <Text, IntWritable>와 같은 Key/Value Pair가 map Function의 output이 되었고 이제 reduce Function의 Input으로 들어가게 된다. reduce Function에서는 이 모든 것을 Key 값을 기준으로 묶고 그 Key가 가지고 있는 output(여기서는 무조건 1이 되겠지)을 더할 예정이다. 그러면 자연스럽게 아래와 같은 결과물이 나오게 된다.

<"I", 2>

<"love", 2>

<"you", 2>

<"do", 1>

<"not", 1>

<"me", 1>

<"am", 1>

<"sad", 1>

결국은 각각의 단어를 세게 되는 것이고, 따라서 WordCount 프로그램이 되는 것이다.

자 이제 마지막으로 이해한 것을 바탕으로 실행시켜보자.

굿굿!! WordCount를 실행시켰다. 결과물도 얻었다. 모두들 잘 이해했기를 바라며 이번 포스팅은 여기서 그만하려한다. 다음 포스팅은 Hadoop Streaming에 대해서 작성할 생각이다.

Reference:

https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapred/JobConf.html

으아 Hadoop, Hadoop이다. 

Big Data 한답시고 이것 저것 정말 시간을 많이 잡아먹고 또 잡아먹는다. 지금까지 내가 한 삽질을 보면 이루말할 것이 없네. 참 그렇다고 새로 막 공부하는 것이라 누구한테 쉽게 물어볼 사람도 없고.. 아무튼 Hadoop도 조금씩 조금씩 공부했었는데 공부한 내용을 조금만 정리해보자.

우선, Hadoop이 무엇인지 소개부터 하련다. Hadoop은 MapReduce 작업을 하기 위한 Open Source Framework이다. Framework는 무엇이냐? 어떤 작업을 하기 위한 프레임이다. 즉, Hadoop으로 우리는 MapReduce를 할 수 있다. (MapReduce가 무엇인지는 http://operatingsystems.tistory.com/entry/MongoDB-MongoDB-Aggregation 이 포스팅에 설명해놨다. MongoDB에서의 MapReduce를 설명해 놓은 링크인데 개념은 같으니 참고하길 바란다.) 우리는 특정 언어를 이용해서 MapReduce 프로그램을 작성할 것이다.(예를 들어, Java) 그 프로그램을 돌릴 때 Hadoop을 사용하여 돌릴 것이다. Java 등의 언어로 쓰여진 MapReduce 프로그램을 쉽게 돌릴 수 있게 해주는 틀, Framework가 바로 Hadoop이다. 

자, 그러면 왜 Hadoop이냐!! MapReduce를 왜 굳이 Hadoop으로 돌리냐? 나름의 장점이 있기 떄문이다.

장점 하나, Accessible. 접근이 용이하다. 여러 대의 Machine에서 Hadoop을 돌릴 수 있다는 것이다. 세계적인 대기업 Amazon에서 Hadoop은 몇 백대의(정확하게 모르겠음) Machine에서 동시에 돌아간다고 한다. (믿거나 말거나) 

장점 둘, Robust. 발생하는 에러 등에 대해서 에러를 핸들링할 수 있다. 이 말은, 다르게 말해서 결과물의 정확도를 위해서 에러가 발생해도 사용자가 올바른 결과물을 가질 수 있도록 해준다는 것이다.

장점 셋. Scalable. 확장성이다. 하나의 Machine을 추가할 때 그냥 Tree에서 Node를 추가하는 것처러므 그저 추가해주면 그만이다. 

장점 넷. Simple. 단순하다. Hadoop을 이용해서 MapReduce를 돌리는 것은 상대적으로 굉장히 단순하게 되어있다. 조금만 익숙해지면 금방 익숙해진다.

여기서 Hadoop의 중요한 특징을 하나 더 말해보자. Distributed 라는 특성을 가지고 있다. 분산 프로그래밍을 가능케 한다. Hadoop을 이용하면, 하나의 MapReduce Job을 금방 끝내게 하기 위해 여러 Machine에 분산시킬 수 있다. 성능이 좋은 하나의 단일 Machine에서 돌리는 것보다 적당한 성능의 Machine 여러대에서 돌리는 것이 더 빠르다는 원리이다.

여기서 하나의 질문을 해보자. 그렇다면 Hadoop은 무조건 여러 대의 Machine이 있어야 돌릴 수 있냐? 아니다!! 단일 Machine에서도 돌릴 수 있다. 그렇지만 이럴 경우 Hadoop의 진정한 장점을 활용할 수는 없겠지. Hadoop은 다음의 세 가지 Mode로 돌릴 수 있다. 

1. Standalone Mode

2. Pseudo-Distributed Mode

3. Fully Distributed Mode

1. Standalone Mode

Standalone Mode는 정말 아무런 추가적인 설정 없이 Single Machine에서 돌릴 수 있는 Mode이다. Hadoop을 Hadoop 공식 홈페이지에서 다운로드 후 압축해제하면 끝이다. Standalone Mode에서는 어떠한 Hadoop 관련 Daemon 돌아가지 않는다. 그저 Framework로만 쓰는 것이다. MapReduce를 실행시킬 때 쓰는 input 또는 output은 Local File System을 이용한다.(다른 Mode처럼 HDFS, Hadoop File System, 등을 쓰지 않는다.) 

Bash> ./bin/hadoop jar ./share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.2.jar grep input output 'dfs[a-z.]+'

위의 명령어는 hadoop으로 jar 파일 안의 grep 이라는 프로그램을 argument 'dfs[a-z.]+'와 실행시키는데 input은 input이라는 폴더 안에, 그 output은 output이라는 폴더에 넣으라는 명령어이다.

input과 output은 ls -al을 수행하면 나오는 실제 local에 존재하는 파일이다. 그래서 local file system이라고 설명한 것이다. 

2. Pseudo-Distributed Mode

Pseudo-Distributed Mode에서는 모든 Hadoop 관련 Daemon이 작동된다. 여기에 추가적으로 Standalone Mode에서 사용하던 Local File System을 사용하지 않고 Hadoop File System, HDFS를 사용하게 된다.

Bash> ./bin/hadoop jar ./share/hadoop/mapreduce/hadoop-mapreduce-examples-2.5.2.jar grep example/input example/output 'dfs[a-z.]+'

나는 개인적으로 hdfs 안에 example이라는 폴더 그 안에 input과 output을 생성해놨다. 이 파일시스템은 Hadoop File System이라서 그냥 shell에서 ls 실행시켜서 볼 수 있는 그런 file system이 아니다.

Bash> ./bin/hdfs dfs -ls /example

위와 같이 hdfs를 통해 ls 명령어를 수행하면 example 밑에 어떤 파일 또는 폴더가 있는지 알수 있다.

Standalone Mode 그리고 Pseudo-Distributed Mode 모두 하나의 Single Machine에서 돌아간다. 따라서 하나의 Machine만 있으면 돌릴 수 있다. 그런데 Hadoop의 Main advantage를 활용하려면 반드시 몇 대 이상의 Machine을 통해 Hadoop을 설치해야하며, 이 때 Hadoop은 Fully Distributed Mode로 작동한다.

3. Fully Distributed Mode

Fully Distributed Mode에서는 여러 대의 Machine을 작업을 분산시킬 수 있다. 물론 여기에서도 Hadoop에서 필요로 하는 모든 Daemon이 작동한다. 

3.1 NameNode and DataNode

3.2 Secondary NameNode

3.3 JobTracker and TaskTracker

3.1 NameNode and DataNode

이전에 말했듯이, Fully Distributed Mode에서는 여러 대의 Machine이 연동되어 동작한다. 각각의 Machine들은 각각의 역할을 가지고 있다. 일단 기본적으로 어떤 Machine은 HDFS를 수행해야 하고 어떤 Machine은 Computation을(연산작업) 수행해야한다. HDFS 그리고 Computation 각각은 Master/Slave 구조로 동작을 하는데, HDFS는 아래와 같은 구조이다.

NameNode와 DataNode는 모두 HDFS 관련 Daemon이며, NameNode가 여러개의 DataNode에 대한 Master이고 DataNode는 Slave이다. 즉, NameNode는 여러 개의 DataNode를 관리하는 역할을 한다.

3.2 Secondary NameNode

만일 NameNode가 어떤 장애로 동작을 못 하게 되면 어떻게 될까? 다른 DataNode들도 모두 작동하지 못 하게 되겠다. 이러면 안 되기 때문에 NameNode의 역할을 대신할 수 있는 Alternative NameNode가 있다. 그것이 Secondary NameNode이다. 

자, HDFS는 이런 식으로 구성하면 되는데, 실제로 작업을 수행하는.. Computation 하는 Machine들은 어떤식으로 구성할까?

3.3 JobTracker and TaskTracker

Computation Machine들 역시 Master/Slave 구조로 동작한다. 각각이 JobTracker 그리고 TaskTracker이다. 여러 개의 TaskTraker들이 작업을 수행하고 그 작업들을 JobTracker가 관리한다. 따라서 아래와 같은 구성을 보일 것이다. JobTracker가 Master 역할, TaskTracker가 Slave 역할을 한다.

후... 이 정도면 Hadoop이 무엇인지 어떤 식으로 동작하는지 개괄적인 내용은 모두 포함한다고 생각한다. 다음 포스팅에서 소개할, 그리고 앞으로 소개할 모든 Hadoop 예제는 Pseudo-Distributed Mode에서 동작한다고 가정하고 진행할 것이다. 다음 포스팅에서는 C 프로그래밍으로 치면 HelloWorld 프로그램, 가장 기본적인 예제 프로그램인 WordCount 예제를 살펴볼 것이다.