map reduce의 입출력과 관계된 이모저모를 기록해본다. 상세한 값들을 다 적으려면 또 인터넷을 뒤져야 하므로 타빨 받쳐주는 수준의 힌트만 적는다. 인터넷에 다 있다.
매퍼의 수는 제한을 걸 수 있는가?
없다. 클러스터 전체에 대한 설정은 가능하지만 매퍼의 수는 입력 파일의 수로 결정된다. 입력 파일은 다시 쪼개서 연산되므로 정확히는 스플릿 수로 결정된다. 다음 설정으로 스플릿이 결정된다. 통상 하둡의 1블럭은 128M나 256M인데 설정을 하지 않으면 이 단위로 자를 것이다.
mapreduce.input.fileinputformat.split.minsize=바이트 수
mapreduce.input.fileinputformat.split.maxsize=바이트 수conf.setNumMapTasks(int num) 즉, mapred.job.maps 를 설정하는 것은 그저 힌팅을 주는 것 뿐이다. 따라서 파일 수가 많으면 많은 매퍼가 뜨는데 다 떠서 자원을 잠식하는 것은 아니다. 아마도 yarn등에 의해서 runnning 갯수를 큐에서 던져주는 방식으로 동작하는 것 같다. FileInputFormatter를 직접 상속 받아 작성하면 이 한계를 넘을 수는 있을 것 같다.
만약 입력 파일의 크기 편차가 크고, 매퍼가 많은 시간을 소모한다면 스플릿을 더욱 줄여야 한다. 대부분의 파일이 다 처리됐는데 256메가 짜리 파일을 처리한다고 매퍼 한 개가 낑낑대고 있을 수 있다. 매퍼가 빨리 끝나는 업무면 괜찮을 수 있것다.
결과 파일의 갯수는 어떻게 되는가?
mapred.job.reduce를 설정하면 결과 파일의 갯수가 된다. 0으로 설정하면 reduce 단계를 생략한다. 따라서 빠르다. 그런데 할당된 매퍼의 수 만큼 파일이 생성된다. 매퍼를 거치면서 필터링되어 결과가 0이라면 0바이트 짜리 파일마저 생긴다. 수백 킬로 짜리 파일들이 양산될 수 있으며 이는 다음 스테이지의 MR등이 받을 때 매퍼 수가 늘어나는 부담이 있다. 리듀스를 일단 타면 하둡 스트리밍의 경우 쏘팅을 무조건 태운다. 쓸데없는 짓인데 생략하는 방법을 못 찾았다.
많은 매퍼 수
매퍼 수가 많아도 클러스터가 알아서 running을 통제하긴 하나 이것을 스케쥴링 하는 것도 일이다. 아몰랑 해서는 안 된다. 줄일 수 있으면 줄여야 한다. 앞단계에 hive가 있다면 merge 를 적절히 하여 작은 파일을 합쳐주는 것이 좋다. merge 옵션은 MR와 TEZ가 각각 존재하니 적절히 맞는 옵션으로 설정해야 한다.
결과 파일의 갯수 줄이기
hive라면 merge 옵션으로 줄일 수 있다. MR이라면 리듀서의 갯수로 줄일 수 있다. 매퍼의 결과 크기나 리듀서의 최종 결과 크기를 예측하여 리듀서의 갯수를 설정하는 것은 현 시점에서 불가능한 것으로 보인다. 다만 FileOutputFormatter를 상속받아 작성해서 적절한 조치를 취할 수는 있다. 이에 대해서 뒤에서 쓴다.
결과 파일의 이름을 내가 원하는 데로 바꾸고 싶다.
FileOutputFormatter를 쓰면 된다. 이를 통해 리듀서를 거치지 않고도 파일 갯수를 통제할 수 있다.
이미 알려진 내용이므로 여기에 다시 가져오면
package com.custom;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
public class CustomMultiOutputFormat extends MultipleTextOutputFormat<Text, Text> {
@Override
protected String generateFileNameForKeyValue(Text key, Text value, String leaf) {
return new Path(key.toString(), leaf).toString();
}
/**
* We discard the key as per your requirement
*/
@Override
protected Text generateActualKey(Text key, Text value) {
return null;
}
}빌드하고 하둡 스트리밍에서 쓰려면,
$JAVA_HOME/bin/javac -cp $(hadoop classpath) -d . CustomMultiOutputFormat.java
$JAVA_HOME/bin/jar cvf custom.jar com/custom/CustomMultiOutputFormat.class
hadoop jar /path/to/your/hadoop-streaming-*.jar -libjars custom.jar -outputformat com.custom.CustomMultiOutputFormat -file your_script.py -input inputpath --numReduceTasks 0 -output outputpath -mapper your_script.py
hadoop 은 아규먼트에 순서가 있으므로 주의한다.
이 때, generateActualKey 메서드를 위 소스와 같이 남겨두면 키가 사라진다.
리듀스 단계를 생략하고 위 코드를 사용하면 어떨까?
위 코드에서 leaf는 MR이 결정한 결과 파일의 이름이다. 리듀스가 생략된다면 넘버링이 붙는다. 리듀서를 거친다면 리듀서의 번호가 붙을 텐데, 이는 키를 이용한 파티셔닝 값이 된다. 아마도 키의 해시값을 나머지 연산으로 퉁쳐서 이 번호가 결정될 것이다. 그런데 leaf를 내 맘대로 설정하면 어찌될까? 데이터가 유실된다.
예를 들어, 위 코드를 통해 파일명을 조작하여 'rrr'로 저장하게 했다고 가정해보자. 키 값에 의해 리듀서 1번과 2번에 각각 들어가야 할 데이터가 전부 'rrr'로 들어간다. 파일이 합쳐지면 다행인데 안 합쳐진다. 관측상으로는 뒤에서 실행되는 리듀서의 데이터만 남는 것 같다. 데이터가 유실된다.
하이브 파티셔닝을 위해 디렉토리 구조로 저장할 수 있을까?
통상적으로 part_hour나 date와 같은 파티셔닝을 많이 쓸 것이다. MR로 떨군 파일을 하이브에 때려넣고 싶은데 파티션을 만들면서 insert overwrite해서는 속도가 매우매우 아햏햏하다. 아예 MR 단계에서 파티션 구조로 저장하는 게 좋다. 위 클래스에서 패쓰를 지정하면 된다. 예를 들면 이렇다. 가져와보니 스플릿이 불필요하게...
return "part_hour" + key.toString().split(",")[0] + "/expire_part_hour=" + key.toString().split(",")[1] + "/" + leaf;파티셔닝 정보에 value가 사용됐다면, 실제 value에서는 없애도 되지 않을까?
그렇다. 파티셔닝 정보에 쓸 값을 키로 쓰는 것이 절약이다. 데이터 줄고 속도도 증가한다. 위에서도 키 값으로 패쓰를 구성했다. 그리고나서 generateActualKey의 null 리턴에 의해 키 값은 이제 무로 돌아간다.
날짜를 키로 쓰고 보니까 리듀서 수를 늘려도 문제가 있는데?
날짜를 키로 쓰면 같은 특정 날짜에만 데이터가 뭉쳐있는 경우 그 파일만 커진다. 분산처리에 좋지 않다. 그런 경우 키 값에 컴마 등으로 구분하여 랜덤값을 넣어준다. 키가 랜덤 % n이라면 n개로 분산되는 것이다. 위의 코드에서 컴마로 스플릿 한 이유가 그것이다.
내 맘대로 결과 파일의 용량 조작하기.
결과 파일의 용량에 일관성이 보인다면 랜덤 % n을 통해 할당된 리듀서가 그 키를 쫓아가도록 하면 된다. % 2로 했을때 100메가 결과 파일이 계속 나온다면 % 4로 하면 50메가 파일이 나올 것이다. 문제는 리듀서의 갯수와 약수 관계가 있어야 한다는 것이다. 그래야 튀는 용량의 파일이 없다. 이유는 설명하기 .. 귀찮 생략..
보통 특정 배치가 들어있는 파티션의 용량만 크다면 그 구간의 키만 % n을 늘리면 된다. 그러면 전체적으로 결과 파일의 크기를 비슷하게 때리 맞출 수 있다.
기술 노트 한 개 남길 때 10분 이상 쓰지 않는다는 규칙에 의해 이상 끗.