hadoop 1.x.x 기준 mapreduce 튜닝 관련

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

== cpu 관련 설정

  - mapred.tasktracker.map.tasks.maximum : 태스크 트래커 최대 병렬 수행 맵 태스크 수

    CPU 총 코어수와 비슷하거나 논리 코어수의 120~150%로 선언.

  - mapred.tasktracker.reduce.task.maxinum : 태스크 트래커 최대 병렬 수행 리듀스 태스크 수

    CPU 총 코어수와 비슷하거나 논리 코어수의 120~150%로 선언.

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

== 디스크 io 관련 설정

  - mapred.compress.map.output : true 인 경우 맵 태스크 결과가 SequenceFile 로 압축되어 네트워크 전송

  - mapred.output.compress : true 인 경우 잡 결과를 압축

  - mapred.map.output.compression.codec : 맵 결과를 압축할 코덱 선택

  - mapred.local.dir : 맵 중간 결과 저장 위치(로컬). 여러 디렉토리 지정시 분산 저장.

  - dfs.data.dir : 데이터 노드가 저장할 블록 위치. 쉼표로 여러 디렉토리 지정시 복제 저장하므로 물리적으로 분리되면 좋음.

  

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

== 메모리 관련 설정

  - mapred.child.java.opts : 태스크 트래커 자식 프로세스 JVM 메모리 크키(-Xmx200m)

  - mapred.child.ulimit : 맵 리듀스 잡 가상 메모리 크기. mapred.child.java.opts 보다 크거나 같게 설정. 2배 이상이 적당. (단위 KB)

  - io.sort.mb : sort 에 사용될 버퍼 크기(단위 MB)

    0.25 * mapred.child.java.opts 보다 크게, 0.5 * mapred.child.java.opts 보다 작게 설정.

  - io.sort.factor : 파일 정렬 시 한번에 병합되는 맵 결과의 파티션 수

    spill, sort, merge 에서 높은 IO 발생하면 값을 높임

  - mapred.job.reduce.input.buffer.percent : reduce 에서 map 결과를 전달받기 위해 사용하는 메모리 크기. 최대 heap 크기의 백분율로 설정

    reduce 단계에서 로컬 디스크 IO 줄이려면 늘림.

    

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

== 네트워크 관련 설정

  - mapred.reduce.parallel.copies : shuffle 단계에서 맵 결과를 가져오는 병렬 전송 수

    맵퍼 태스크의 결과가 많은 경우 설정 증가 시킴. 단 CPU 사용률 증가.

  - topology.script.file.name : DNS 를 네으쿼트 토폴로지 명으로 변환할 스크립트 지정

    rack awareness. 미설정시 같은 rack으로 간주함

    

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

== 그 외 hdfs-site.xml 주요 설정

  - dfs.access.time.precision : 파일 접근 시간 정밀도(단위 ms). 0 인 경우 접근시간 기록하지 않음 -> namenode 로그 기록 성능 향상

  - dfs.balance.bandwidthPerSec : 데이터노드간 최대 대역폭(byte/sec)

  - dfs.block.size : 신규 생성될 파일의 기본 block 크기. 입력 파일이 큰경우 늘림.

  - dfs.datanode.du.reserved : block 을 저장하는 각 경로의 최소 확보 공간 지정

  - dfs.datanode.handler.count : block 요청 처리 스레드 수. 데이터노드 처리율을 높일 수 있음. dfs.data.dir 가 여러 디렉토리인 경우 성능 향상.

  - dfs.max.objects : 최대 object 수(file, directory, block). 0은 무제한.

  - dfs.name.dir : namenode의 block 저장 경로. 쉼표로 여러 디렉토리 지정시 라운드 로빈으로 저장하며 물리적으로 분리되면 좋음.

  - dfs.name.edits.dir : namenode의 트랜잭션(edits) 파일 저장 위치. 쉼표로 여러 디렉토리 지정시 복제.

  - dfs.namenode.handler.count : namenode 스레드 수

  - dfs.replication : 파일 저장 시 데이터 노드 복제 수

  - dfs.replication.considerLoad : true 지정 시 데이터 복제 대상 노드 선택시 데이터 노드의 부하 고려

  

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

== 그 외 core-site.xml 주요 설정

  - fs.default.name : 파일 시스템 이름(URI) 지정

  - hadoop.tmp..dir : 임시 폴더 지정

  - fs.checkpoint.dir : secondary namemode 체크포인트 저장 경로. 쉼표로 여러 디렉토리 지정시 복제.

  - io.file.buffer.size : 읽기/쓰기 작업에 사용되는 버퍼 크기.

  

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

== 성능 모니터링 툴

  - 툴 : checkwa, ganglia, nagios, ambari

  - 리눅스 : dstat, top, htop, iotop, vmstat, sar, netstat, perf, strace

  - 메모리 : vmstat 의  so(메모리에서 디스크로 스왑), si(디스크에서 메로리로 스왑) 항목

  - 디스크 : TestDFSIO 벤치마크 툴 

  - 네트워크 : 최대 대역폭의 1/8에 근접하면 과부하

    데이터 수송률 : dstat --nf

    인터럽트율 : dstat -i, dstat -if

    

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

== 성능 모니터링

  

  - 노드 : blacklisted, graylisted, excluded nodes 확인 (웹 페이지)

  - 입력 데이터 크기 : HDFS_BYTES_WRITTEN, Reduce shuffle bytes, Map output bytes, Map input bytes (웹 페이지)

  - 대규모 IO, Network 트래픽 : FILE_BYTES_READ, HDFS_BYTES_READ, Bytes Written, HDF_BYTES_WRITTEN (웹 페이지)

  - 병렬 태스크 부족 : Map Task Capacity, reduce Task Capacity, Num Tasks, Running (웹 페이지)

  - CPU : vmstat 의 in(초당 인터럽트 호출 수 : 서버 과부하 혹은 장비 문제), cs (context switching)

    vmstat -S M 1 : 1초마다 출력

    

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

== 하둡 클러스터 규모 산정

  * 하둡은 복제 하므로 RAID 불필요. HDFS 외의 용도로 30%의 하드 여유공간 권장.

  * 용량에 따른 노드 수 산정

    일별 입력 데이터 스토리지 사용량 = 일별 입력 데이터(100GB) * 복제수 (3) = 300GB

    월별 데이터 크기 = (일별 입력 데이터 스토리지 사용량(300GB) * 한달(30)) * 1.05(월별 증가량 5%인 경우) = 9450GB

    1년 후 데이터 크기 = 9450GB * (1+0.05)^12 = 16971GB

    할당 곤간 = HDD크기(4TB) * (1 - (맵리듀스 중간 데이터 / 100 + 디스크별 HDFS 외 예약된 공간 / 100))

      = 4(TB) * (1- (0.25 + 0.30)) = 1.8TB (노드 최대 용량)

    ∴ 첫번째 월별 데이터 = 9450 gb/ 1800gb ≒ 6 노드

    ∴ 12번째 월별 데이터 = 16971 gb/ 1800gb ≒ 10 노드

  * 마스터 노드 메모리 산정

    메모리 크기 = HDFS 클러스 관리 메모리(2 ~ 4 gb) + namenode 메모리(2 ~4 gb) + OS 메모리 (4 ~ 8gb)

    최소 namenode 메모리 = 2 + 2 + 4 = 8 (secondary namenode 동일)

    

  * 슬레이브 노드 메모리 산정

    메모리 크기 = cpu 코어별 메모리(4 ~ 8gb) * cpu 코어수(4 ~) + datanode 프로세스 메모리(4 ~ 8gb) + task tracker 프로세스 메모리(4 ~ 8gb) + OS 메모리(4 ~ 8gb)

    최소 datanode 메모리 = 4 * 4 + 4 + 4 + 4 = 28gb

    

  * 네트워크

    이더넷 인트라-랙과 N X 10 GB 이더넷 인터-랙 추천

    

  * map reduce 수

    대규모 데이터 작업 시 각 노드의 최대 가용 맵퍼 수 : 물리 코어 수 - 예약된 코어 * (0.95 ~ 1.5)

    예약된 코어 = task tracker 1 코어 + datanode 1코어

    ex)

      노드의 cpu를 하이퍼스레딩으로 120% 활용 시 최대 가용 맵퍼 수 = 8 - 2 * 1.2 = 7.2 = 7 (반올림)

      reducer 를 mapper의 2/3 설정 시 최대 가용 reducer 수 = 7 * 2/3 = 5

      ∴ mapper = 7 * 40(클로스터 노드 수) = 280 = mapred.tasktracker.map.tasks.maximum

      ∴ reducer = 5 * 40 = 200 = mapred.tasktracker.reduce.task.maxinum

    

      

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

== map 성능 향상

  - 실행 순서 :

    -- read(입력) : 입력 데이터 크기 측정 (dfs.block.size 설정)

    -- map : 쏠림현상 감시(데이터 파일이 대량 입력되는지, 분할 안되는 큰 파일이 입력되는지)

      avro 로 컨테이너 파일 생성.

      har 하둡 아카이브 파일 사용. (hadoop archive 명령)

      SequenceFile 로 파일 합치기

      * 데이터 분할 미지원 : deflate, gzip, lzo, snappy 

    -- spill : spill 빈도 줄이기

      combiner 사용

      설정 수정 : 

        레코드 길이(RL) = 맵 결과 바이트 수 / 맵 결과 레코드 수

        spill 된 레코드 크기(RS) = spill 된 레코드 수 * 레코드 길이(RL)

        메타데이터 크기(MS) = 메타데이터 길이 * spill 된 레코드 수

        io.sort.mb = 메타데이터 크기 + spill 된 레코드 크기. 슬레이브 노드의 가용 메모리 자원 1/5의 70~75% 수준으로 정의.

        io.sort.record.percent = 메타데이터 길이 / (메타데이터 길이 + 레코드 길이). 초기값 권장

        io.sort.factor = io.sort.mb 의 1/10

        io.sort.spill.percent = 초기값 권장. 버퍼 영역 최대 활용시 0.99 설정.

        tasktracker.http.threads = task tracker 별 설정. map 결과를 reduce로 전달할 때 사용하는 스레드 수.

        

    -- fetch(취합) : 메모리 버퍼링 시간, 중간 데이터 크기 측정.

    -- merge(병합) : reduce 로 전달하기 위해 spil 파일을 하나로 merge 하는 시간 츶겅.

    

  - map 처리율 = SUM(각 map input bytes) / SUM(각 map 작업 시간 sec)

  평균 map 처리율 = map 처리율 / map task 수

  

  

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

== reduce 성능 향상

  - 실행 순서 :

    -- shuffle : map 중단 데이터를 reduce task 로 전송 및 전송 받은 데이터 sort/merge 시간 측정

    -- reduce : 작업 시간 측정. 입렉 데이터 크기 확인.

    -- output : 결과를 hdfs 에 쓰는 시간 측정.

    

  - reduce 처리율 = SUM(각 reduce shuffle bytes) / SUM(각 reduce 작업 시간 sec)

  - 설정 수정 : 

    -- mapred.reduce.parallel.copies : reduce task 스레드 수

    -- io.sort.factor

    -- mapred.job.reduce.input.buffer.percent : reduce 단계에서 map 결과 보관에 필요한 메모리를 전체 heap 메모리 크기의 비율로 선언.

    -- mapred.job.shuffle.input.buffer.percent : shuffle 단계에서 map 결과 보관에 필요한 메모리를 전체 heap 메모리 크기의 비율로 선언.

    -- tasktracker.http.threads

    

    

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

== 압축 사용 - 네트워크 전송량 축소 및 disk 효율적 사용

  - 압축 적용 가능한 단계 : 

    -- map : 입력 데이터 압축

      분할 가능 코덱 사용(zlib, bzip, SequenceFile, ..)

    -- shuffle/sort : map 결과 압축

      빠른 코덱 사용(LZO, LZ4, Snappy)

    -- reduce : reduce 결과 압축

      reduce 결과가 보관용이면 gzip, bzip2

      다른 mapreduce 입력용이면 빠른 코덱 사용

      

  - 관련 설정 : 

    -- o.compression.codec : 입력 데이터 압축 코덱 설정.

    -- mapreduce.map.output.compress : mapper 결과 압축 사용 여부

    -- mapreduce.map.output.compress.codec : mapper 결과 압축 코덱 설정

    -- mapreduce.output.fileoutputformat.compress : reducer 결과 압축 사용 여부

    -- mapreduce.output.fileoutputformat.compress.codec : reducer 결과 압축 코덱 설정

    -- mapreduce.output.fileoutputformat.compress.type : reducer 결과 압축 형식(SequenceFile 결과 압축 혁식으로 NONE, RECORD, BLOCK 중 하나)

    

  - 그 외 : 

    -- 적절한 Writable 자료형 사용으로 shuffle, sort 시 속도 향상. 아래 2가지 비교방식 존재.

      - org.apache.hadoop.io.WritebleComparable 인터페이스 구현

      - RawComparator 구현 : 원시적 바이트 비교방식으로 WritebleComparable 보다 성능 향상 이점.

    -- InputFormat 인터페이스 :

      mapper로 데이터 전달 방식/출처 정의. 입력된 파일을 FileInputSplit 인스턴스 조각으로 분할(이는 mapper 입력으로 전달)

      InpurFormat 보다 FileInputFormat 상속 권장.

    

    -- RecordReader 인터페이스 :

      입력 파일로 부터 레코드 읽는 방식/ mapper로 key/value 전달 방식 정의

    

    -- 객체 new 줄이기

    

    -- 튜닝 목표 : 더 빠르게, 더 적은 자원, 더 많은 결과

    

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

== HW 튜닝, OS 권장 설정

  - 하둡 클러스터 체크 리스트

    클러스터 모든 모드 간 통신 가능 확인.

    관리자가 모든 노드의 물리적/원격 접속 관리 가능 여부 확인.

    서비스당 최소 하나의 노드 장애 허용 여부 확인.

    비사용 장비 및 랙 공간 확보 - 서버 호스팅 변수 등 클러스터 환경 한계 확인

    장애 발생시 고가용성을 위한 비상 대책 정의.

    하둡 스토리지 공간 최대 활용을 위한 데이터 위치 및 백업 계획 정의

    

  - BIOS 튜닝 체크 리스트

    모든 CPU 코어 활성화 확인.

    NCQ(native command queuing mode) 모드 활성화 - io 성능 향상. 주로 biso AHCL(advanced host controller interface) 에서 설정.

    

  - OS 권장 설정

    ext4 파일 시스템 사용. (리눅스 배포판 선택 시 고려)

    noatime 설정 해지 - 디스크의 모든 읽기 작업시 마지막 접근시간 기록 해지

    nodiratime 설정 해지 - 디렉토리 조회 시 접근 시간 기록 해지

    LVM(logical volumne management) 비권장 - disk io 성능 저하 문제.

    리눅스 swap 메모리 크기 최소화 - 커널 swqp 최소화.

    CFQ(completely fair queuing) IO 스케쥴러 사용 - IO 작업이 원형 큐에서 처리되는 라운드 로빈 알고리즘 사용하고 각 IO 작업에 정해진 작업 시간이 주어짐.

    리눅스에서 동시 작업 가능한 파일 디스크립터 숫자 늘려 mapreduce 성능 향상.

    

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

== 하둡 튜닝(권장)

  - 메모리 설저 조정 - OOM 발생 회피

  - JVM 재사용 활성화

  - JVM 코드 캐시 부족시 확장

  - GC 로그 정밀 분석 및 하둡 힙 사용률 확인

  - 압축 사용

    대규모 입력 데이터의 경우 압축 사용

    disk IO 최소화를 위한 map 결과 압축

    combinaer 사용

    job 결과 압축 : mapred.compress.map.output, mapred.output.compression.type

  - 대량의 spill 발생 시 io.sort.mb, io.sort.record.percent, io.sort.spill.percent 설정 최적화.

  - 복제수 조정

  

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

== 하둡 사용을 위해 정의해야할 사항

  - 설치/관리 필요한 컴포넌트 정의

  - 설치 방식 선택(수동, peppet, yum, ...)

  - 스토리지 기술 선택(hdfs, hbase, ...)

  - block 크기, 복제수, 압축 여부 등 정의

  - 모니터링 방식 정의

  - 자원 산정 (disk 용량, mapreduce 를 위한 필수 노드 수, ram, cpu, network 대역폭, ..)

  - map/reduce 태스크 수 산정

  - 가상환경은 개발환경에서만 사용. (상용 환경에선 피해야 함)

  - map 의 spill 과 reduce 의 disk IO 최소화.

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

== mapreduce 작성시 고려사항

  - 결과 파일에 큰 경우 dfs.block.size 설정 조절(늘린다)

  - 압축 속도/압축률 고려하여 압축 사용

  - 하나 이상의 결과 파일 생성 비권장

  - 결과 파일 포멧의 적절한 선택

  - 대규모 압축 텍스트 데이터를 쓰기 위해 분할 압축 코덱만 사용

    (zlib,gzip,lzo는 분할 처리가 안되어 단일 map이 모든 입력을 처리하는 상황 발생 하므로 압축/분할 처리 가능한 SequenceFile 고려)