jaso extends j2ee

hadoop에서 MapTask에서 map function의 결과를 reduce로 보내기 위해서는 map function에서 collector.collect()를 호출한다. 하지만 collect() 메소드를 호출하는 즉시 파일시스템으로 저장하지 않는다. 메모리 버퍼에 계속 쌓아 놓은 다음 일정 사이즈가 되면 메모리에 있는 내용을 key를 이용하여 정렬한 다음에 Combiner가 지정되어 있으면 combiner로 보내고 그렇지 않으면 파일시스템으로 저장한다.
환경 변수 중 "io.sort.mb" 값이 메모리 버퍼의 크기를 설정하는 값이다. 기본은 100MB로 되어 있다. 따라서 MapTask에서 필요한 메모리를 계산하기 위해서는 io.sort.mb 값까지 같이 고려해야 한다. 예를 들어 하나의 map task에서 약 200MB 정도의 메모리를 사용할 것으로 예상되는 경우라면 mapred.child.java.opts에서는 300MB 이상으로 설정해야 한다.

여기서 고려해야 될 사항이 하나 더 있다. http://www.jaso.co.kr/252 에서 처럼 자바의 경우 실제 메모리 사용량보다 더 많은 메모리를 사용하고 있기 때문이다. 자바에서 메모리 사용량을 정확하게 계산하는 것은 어렵다.
Hadoop의 경우 io.sort.mb 값과 메모리 사용량을 비교해야 하는데 이 부분을 어떻게 처리하는지 참고하면 메모리 사용량을 찾을 것 같아서 소스를 좀 더 찾아 보았는데 0.16 버전과 0.17 버전에서 다르게 처리하고 있는 것을 알 수 있었다.
0.16에서는 BasicTypeSorterBase 클래스의 getMemoryUtilized() 메소드에 구현되어 있다.

"(startOffsets.length) * BUFFERED_KEY_VAL_OVERHEAD +
              maxKeyLength + maxValLength"

이처럼 구현되어 있는데 startOffsets.length는 저장된 값의 갯수이고 maxKeyLength 는 저장된 key의 max 길이 값, maxValLength는 저장된 value의 max 길이 값이다. BUFFERED_KEY_VAL_OVERHEAD 는 다음과 같은 주석이 달려 있다.

  //the overhead of the arrays in memory
  //12 => 4 for keyoffsets, 4 for keylengths, 4 for valueLengths, and
  //4 for indices into startOffsets array in the
  //pointers array (ignored the partpointers list itself)
  static private final int BUFFERED_KEY_VAL_OVERHEAD = 16;

하나의 key, value 값을 저장하기 위해 16byte의 오버헤드가 더 추가 된다는 것을 알 수 있다.

0.17에서는 몇가지 옵션 값이 더 추가 되었다. "io.sort.record.percent",  "io.sort.spill.percent" 값이다.
io.sort.record.percent는 버퍼에 저장된 레코드의 갯수가 일정 수준에 도달하면 flush() 시키기 위한 값이다. io.sort.mb * 1024 * 1024 * io.sort.record.percent / 4 보다 갯수가 많으면 flush() 한다.
io.sort.spill.percent는 io.sort.mb 값에 100% 도달했을 때 저장하는 것이 아니라 버퍼의 메모리 용량이  io.sort.spill.percent 에 도달하면 flush() 하게 된다.

Map Task 수행 중 가끔 OutOfMemory 가 발생하는 경우가 있는데 이때에는 mapred.child.java.opts 값을 무작정 늘리기 보다는 위의 값들에 대한 처리도 적절하게 해야 한다.

크리에이티브 커먼즈 라이센스

이 저작물은 크리에이티브 커먼즈 코리아 저작자표시-비영리-변경금지 2.0 대한민국 라이센스에 따라 이용하실 수 있습니다.

Posted by 김형준