MapReduce 실행 절차

MapReduce 에는 다음과 같은 네 개의 개별 솔리드가 있습니다.

1.JobClient: 클라이언트 노드에서 실행되며 MapReduce 프로그램을 Jar 패키지로 변환하여 HDFS 에 저장하고 Jar 패키지의 경로를 Jobtracker 에 제출하고 job tracker 가 작업을 배포 및 모니터링합니다.

2.JobTracker: name node 에서 실행되고, JobClient 가 제출한 작업을 수신하고, TaskTracker 에서 실행되도록 작업의 개별 하위 작업을 예약하고, 모니터링합니다. 실패한 작업이 발견되면 다시 실행됩니다.

3.TaskTracker: 데이터 노드에서 실행되고, JobTracker 와 능동적으로 통신하고, 작업을 받고, 각 작업을 직접 수행합니다.

4.HDFS: 작업 파일을 다른 엔티티와 공유하는 데 사용됩니다.

자세한 내용은 다음과 같습니다.

1.JobClient 는 RPC 프로토콜을 통해 JobTracker 에 새 응용 프로그램의 ID 를 요청하여 MapReduce 작업의 ID 를 가져옵니다.

2. 작업 트래커는 작업의 출력 설명을 확인합니다. 예를 들어 출력 디렉토리가 지정되지 않았거나 디렉토리가 이미 있는 경우 작업이 제출되지 않고 작업 클라이언트에 오류가 발생합니다. 그렇지 않으면 새 작업 ID 가 작업 클라이언트에 반환됩니다.

3. 작업 클라이언트는 작업 JAR 파일, 구성 파일 및 계산된 입력 조각을 포함하여 작업에 필요한 리소스를 작업 ID 로 명명된 HDFS 폴더에 복사합니다.

4. 작업 클라이언트는 submitApplication () 을 통해 작업을 제출합니다.

5.JobTracker 는 이를 호출하는 submitApplication () 메시지를 받은 후 작업을 초기화합니다.

6.JobTracker 는 HDFS 에서 처리할 파일을 읽고 각각 TaskTracker 에 해당하는 입력 슬라이스 계산을 시작합니다.

지도 작업은 작업 추적자에게 무작위로 할당되지 않습니다. 여기 데이터 로컬이라는 개념이 있습니다. 즉, 지도 작업을 지도에 의해 처리된 데이터 블록이 포함된 TaskTracker 에 할당하고, 동료는 프로그램 jar 패키지를 TaskTracker 에 복사하여' 동작, 데이터 고정' 이라고 합니다. Reduce 작업을 할당할 때 데이터 현지화를 고려하지 않았습니다.

7.TaskTracker 는 하트비트 메커니즘을 통해 작업 (작업에 대한 설명 정보) 을 수신합니다.

8.TaskTracker 는 작업 리소스 (JAR 패키지, 프로필 등) 를 읽습니다. ) HDFS 에 있습니다.

9.TaskTracker 는 특정 작업 (MapperTask 또는 ReducerTask) 을 수행하는 Java 하위 프로세스를 시작합니다.

10.TaskTracker 는 계약 결과를 HDFS 에 기록합니다.

참고: 하나의 HDFS 블록 크기 (기본값 64M) 를 슬라이스로 사용하시겠습니까?

15 버전 2.7.3 이후 블록 크기가 64 MB 에서 128 MB 로 변경되었습니다.

셔플 분석

Shuffle 의 중국어는' 셔플' 을 의미한다. 이렇게 보면, map 생성 데이터가 hash 프로세스를 통해 서로 다른 reduce 작업에 할당됩니다. 이는 셔플 데이터 프로세스입니까? (윌리엄 셰익스피어, Northern Exposure (미국 TV 드라마), 성공명언)

-응? 셔플 개념:

Collections.shuffle (목록 목록): 목록에서 요소의 순서를 무작위로 뒤섞습니다.

MapReduce 의 Shuffle: 지도 작업에서 데이터를 내보내고 작업 입력을 줄이는 과정을 설명합니다.

매핑 종료 프로세스 분석

1 각 입력 조각은 하나의 맵 작업에 의해 처리됩니다. 기본적으로 HDFS 의 블록 크기 (기본값 64M) 는 세그먼트로 사용되지만 블록 크기도 설정할 수 있습니다. 맵 출력의 결과는 일시적으로 루프 메모리 버퍼에 배치됩니다 (버퍼 크기는 기본적으로 io.sort.mb 속성에 의해 제어되는 100M 으로 설정됨). 버퍼가 오버플로우될 때 (기본적으로 버퍼 크기의 80%, io.sort.spill.percent 등록 정보에 의해 제어됨) 로컬 파일 시스템에 오버플로우 파일이 생성됩니다.

디스크에 쓰기 전에 스레드는 reduce 작업 수에 따라 데이터를 동일한 수의 파티션으로 나눕니다. 즉, reduce 작업당 하나의 파티션에 해당하는 데이터입니다. 이는 일부 reduce 작업에 많은 양의 데이터가 할당되지 않도록 하기 위한 것이며, 일부 reduce 작업에는 데이터가 거의 또는 전혀 할당되지 않는 어색한 상황을 방지하기 위한 것입니다. 실제로 파티션은 데이터를 해시하는 프로세스입니다. 그런 다음 각 구역의 데이터를 정렬합니다. 이때 Combianer 가 설정되어 있으면 정렬된 결과가 병합됩니다. 이렇게 하는 목적은 가능한 적은 양의 데이터를 디스크에 쓰는 것입니다.

지도 작업이 마지막 레코드를 출력할 때 오버플로우 파일이 많을 수 있으므로 병합해야 합니다. 병합 중 정렬 및 병합 작업은 두 가지: 1 을 사용하여 디스크에 쓸 때마다 데이터 양을 최소화합니다. 2. 다음 복제 단계에서는 네트워크에서 전송되는 데이터의 양을 최소화합니다. 마지막으로 하나의 파티션과 정렬된 파일로 결합됩니다. 네트워크에서 전송되는 데이터의 양을 줄이기 위해 mapred.compress.map.out 을 true 로 설정하면 데이터를 압축할 수 있습니다.

데이터 압축: Gzip, Lzo, snappy.

4 파티션의 데이터를 해당 reduce 세션으로 복사합니다. 누군가는' 구역의 데이터가 어떤 reduce 에 해당하는지 어떻게 알 수 있을까?' 라는 질문을 할 수 있습니다. 사실 지도 임무는 아버지 TaskTracker 와 계속 연락하고, TaskTracker 도 obTracker 와 계속 심장 박동을 유지한다. 따라서 전체 클러스터에 대한 매크로 정보는 작업 트래커에 저장됩니다. Reduce 작업은 JobTracker 에서 해당 지도 출력 위치만 가져오면 됩니다.

최종 프로세스 분석 감소

1 reduce 는 서로 다른 지도 작업에 대한 데이터를 받으며, 각 지도의 데이터는 질서 정연합니다. Reduce 측에서 수신하는 데이터의 양이 상당히 적은 경우 메모리에 직접 저장됩니다 (버퍼 크기는 이 용도로 사용되는 힙 공간의 백분율을 나타내는 mapred.job.shuffle.input.buffer.percent 속성에 의해 제어됨). 데이터 양이 버퍼 크기의 일정 비율 (mapred.job.shuffle.merg.percent 에 의해 결정됨) 을 초과하면 데이터가 병합되어 디스크로 넘칩니다.

오버플로 파일이 증가함에 따라 백그라운드 스레드는 이후 병합을 위한 공간을 절약하기 위해 더 크고 질서 있는 파일로 결합합니다. 실제로 Mapreduce 는 map 과 reduce 에서 정렬 및 병합 작업을 반복합니다. 이제 누군가가 서열이 Hadoop 의 영혼이라고 말하는 이유를 이해하게 되었다.

3 병합 중에는 중간 파일 (디스크 쓰기) 이 많이 있지만 MapReduce 를 사용하면 디스크에 기록되는 데이터가 최소화되고 최종 병합 결과는 디스크에 기록되지 않고 Reduce 함수에 직접 입력됩니다.

4 감속기 입력 파일. 지속적인 통합을 통해 결국 "최종 파일" 이 생성됩니다. 왜 따옴표를 넣어야 합니까? 이 파일은 디스크나 메모리에 있을 수 있기 때문입니다. 우리에게는 Reducer 의 입력으로 메모리에 저장하고자 하지만, 기본적으로 이 파일은 디스크에 저장됩니다. Reducer 의 입력 파일이 설정되면 전체 셔플이 끝납니다. 그런 다음 Reducer 를 실행하고 결과를 HDSF 에 배치합니다.

세부 프로세스:

1? 각 지도 작업에는 지도의 출력을 저장하는 메모리 버퍼가 있습니다. 버퍼가 거의 가득 차면 버퍼의 데이터를 디스크에 임시 파일로 저장해야 합니다. 전체 지도 작업이 완료되면 map 작업이 디스크에서 생성한 모든 임시 파일을 병합하고 최종 공식 출력 파일을 생성한 다음 reduce 작업이 데이터를 당길 때까지 기다립니다.

2? 지도 작업을 수행할 때 입력 데이터는 HDFS 블록에서 가져옵니다. 물론 MapReduce 개념에서 map task 는 split 만 읽습니다. Split 과 block 사이의 대응 관계는 다대일 수 있으며 기본값은 일대일입니다. Wordcount 예제에서는 지도의 입력 데이터가 모두 "AAA" 와 같은 문자열이라고 가정합니다.

3? Mapper 의 연산으로 우리는 mapper 의 출력이 키/값 쌍이라는 것을 알고 있습니다. 키는 "AAA" 이고 값은 1 입니다. 현재 map end 는 1 의 덧셈만 수행하므로 결과 세트는 reduce 세션에서 통합됩니다. 이전에는 이 작업에 세 가지 reduce 임무가 있다는 것을 알고 있었습니다. 그렇다면 현재의' AAA' 는 어떤 축소를 처리해야 할까요? 너는 지금 결정을 내려야 한다.

4? MapReduce 는 키 또는 값의 수와 Reduce 에 따라 현재 출력 데이터를 처리해야 하는 reduce 작업을 결정하는 Partitioner 인터페이스를 제공합니다. 기본적으로 키 해시 뒤에는 reduce 작업 데이터가 있는 모듈 (나머지) 이 옵니다. 기본 모드는 reduce 의 처리 능력을 평균화하는 것입니다. 사용자가 파티션에 대한 요구 사항이 있는 경우 이를 사용자 정의하고 작업에서 설정할 수 있습니다.

5? 이 예에서 "AAA" 는 파티션 뒤에 0 을 반환합니다. 즉, 이 값 쌍은 첫 번째 reduce 에 의해 처리되어야 합니다. 다음으로 메모리 버퍼에 데이터를 써야 합니다. 버퍼는 지도 결과를 대량으로 수집하여 디스크 IO 의 영향을 줄이는 데 사용됩니다. 키/값 쌍 및 파티션 결과가 버퍼에 기록됩니다. 물론 키 값은 쓰기 전에 바이트 배열로 직렬화됩니다.

6? 메모리 버퍼의 크기가 제한되어 있으며 기본값은 100MB 입니다. 지도 작업 출력이 많은 경우 메모리가 폭주할 수 있으므로 버퍼의 데이터를 특정 조건에서 디스크에 임시로 기록한 다음 다시 사용해야 합니다. 메모리에서 디스크로 데이터를 쓰는 이 과정을 오버플로우라고 하며, 중국어는 오버플로우로 해석될 수 있다. 재정의는 별도의 스레드에 의해 이루어지며 매핑 결과를 버퍼에 기록하는 스레드에는 영향을 주지 않습니다. 오버플로 스레드가 시작될 때 맵 결과의 출력을 차단하지 않아야 하므로 전체 버퍼의 오버플로 비율은 spill.percent 이고 기본 비율은 0.8 입니다. 즉 버퍼의 데이터 값이 임계값에 도달한 경우입니다 (버퍼 크기 * spill percentage =1; 지도 작업의 출력도 나머지 20MB 메모리에 쓸 수 있으며 서로 영향을 주지 않습니다.

7? 오버플로우 스레드가 시작되면 이 80MB 공간에서 키를 정렬해야 합니다. 정렬은 MapReduce 모델의 기본 동작이며, 여기서 정렬은 직렬화된 바이트의 정렬입니다.

8? 맵 작업의 출력을 다른 reduce 터미널로 보내야 하고 메모리 버퍼가 동일한 reduce 터미널로 보낼 데이터를 병합하지 않기 때문에 이 병합은 디스크 파일에 반영되어야 합니다. 관도에서 볼 수 있듯이 디스크에 기록된 일부 파일은 이미 다른 reduce 터미널의 값을 병합했습니다. 따라서 오버플로우 프로세스의 중요한 세부 사항은 많은 키/값 쌍을 reduce 터미널로 보내야 하는 경우 분할 영역과 관련된 인덱스 레코드를 줄이기 위해 이러한 키/값을 결합해야 한다는 것입니다.

각 reduce 측의 데이터를 통합할 때 일부 데이터는 "AAA"/1,"AAA"/1과 같이 보일 수 있습니다. Wordcount 예제의 경우 단어 발생 횟수를 간단히 계산합니다. 동일한 지도 작업의 결과에 "AAA" 와 같은 키가 여러 번 나타나면 해당 값을 결합해야 합니다. 이 과정을 감소 또는 병합이라고 합니다. 그러나 MapReduce 용어에서 Reduce 는 계산을 위해 여러 maptasks 에서 데이터를 가져오는 프로세스만 수행합니다. Reduce 를 제외하고 비공식 통합 데이터는 통합으로만 계산할 수 있습니다. 사실, MapReduce 는 병합기를 복원기와 동일시한다는 것을 모두 알고 있다.

클라이언트에 이미 통합기가 설정되어 있다면 지금은 통합기를 사용할 때입니다. 동일한 키를 가진 키/값 쌍의 값을 더하여 디스크로 넘친 데이터의 양을 줄입니다. Combiner 는 MapReduce 의 중간 결과를 최적화하므로 모델 전체에서 여러 번 사용됩니다. 그렇다면 통합자를 사용할 수 있는 장면은 무엇입니까? 이 분석에서 통합기의 출력은 계약자의 입력이며, 통합기는 최종 계산 결과를 변경할 수 없습니다. 그래서 제 생각에는 Combiner 는 Reduce 의 입력 키/value 와 출력 키/value 가 정확히 같은 장면에만 사용해야 하며 최종 결과에는 영향을 주지 않습니다. 누적, 최대값 등. 콤비네이터는 반드시 조심해서 사용해야 한다. 잘 사용하면 작업 실행의 효율성에 도움이 됩니다. 그렇지 않으면 reduce 의 최종 결과에 영향을 줍니다.

9? 오버플로우할 때마다 디스크에 오버플로우 파일이 생성됩니다. 맵의 출력이 정말 크고 이러한 오버플로가 많은 경우 디스크에 오버플로 파일이 여러 개 있을 수 있습니다. Map 작업이 실제로 완료되면 메모리 버퍼의 모든 데이터가 디스크로 넘쳐 오버플로우 파일을 형성합니다. 결국 디스크에 이러한 오버플로우 파일이 하나 이상 있습니다. map 의 출력이 적은 경우 map 을 실행할 때 오버플로우 파일이 하나만 생성됩니다. 마지막으로 파일이 하나뿐이므로 이러한 오버플로우 파일을 병합해야 합니다. 이 프로세스를 병합이라고 합니다. 합병은 어떤 건가요? 앞의 예와 마찬가지로 "AAA" 는 한 시간 값이 5 인 맵 작업에서 읽고 다른 시간 값이 8 인 맵에서 읽습니다. 같은 키를 가지고 있기 때문에 하나의 그룹에 병합해야 합니다.

그룹이란 무엇입니까? "AAA" 의 경우 실제 태양: {"AAA", [5,8,2, ...]}, 다른 오버플로우 파일에서 배열의 값을 읽고 추가합니다. Merge 는 여러 오버플로우 파일을 하나의 파일로 결합하기 때문에 동일한 키가 있을 수 있습니다. 이 과정에서 클라이언트가 병합기를 설정하면 병합기를 사용하여 동일한 키를 병합합니다.

이 시점에서 지도측의 모든 작업이 완료되었으며 결과 파일도 TaskTracker 가 도달할 수 있는 로컬 디렉토리에 저장됩니다. 한 번 줄일 때마다? Task 는 JobTRacker 로부터 map 작업이 RPC 를 통해 완료되었는지 여부에 대한 정보를 지속적으로 입수하고 있습니다. Reduce task 가 TaskTracker 에서 map 작업 실행이 완료되었다는 통지를 받으면 Shuffle 프로세스의 후반부가 시작됩니다.

감속 측면의 셔플 프로세스:

1 복제 프로세스, 간단한 데이터 풀. Reduce 프로세스는 biner 작업을 통해 1, 디스크에 쓸 때마다 데이터 양을 최소화하는 두 가지 목적으로 일부 데이터 복제 스레드 (Fetcher) 를 시작합니다. 2. 다음 복제 단계에서는 네트워크에서 전송되는 데이터의 양을 최소화합니다. 마지막으로 하나의 파티션과 정렬된 파일로 결합됩니다. 네트워크에서 전송되는 데이터의 양을 줄이기 위해 mapred.compress.map.out 을 true 로 설정하면 데이터를 압축할 수 있습니다.

데이터 압축: Gzip, Lzo, snappy.

4 파티션의 데이터를 해당 reduce 세션으로 복사합니다. 누군가는' 구역의 데이터가 어떤 reduce 에 해당하는지 어떻게 알 수 있을까?' 라는 질문을 할 수 있습니다. 사실 지도 임무는 아버지 TaskTracker 와 계속 연락하고 있고, TaskTracker 도 obTracker 와 계속 심장 박동을 유지하고 있다. 따라서 전체 클러스터에 대한 매크로 정보는 작업 트래커에 저장됩니다. Reduce 작업은 JobTracker 에서 해당 지도 출력 위치만 가져오면 됩니다.

최종 프로세스 분석 감소

1 reduce 는 서로 다른 지도 작업에 대한 데이터를 받으며, 각 지도의 데이터는 질서 정연합니다. Reduce 측에서 수신하는 데이터의 양이 상당히 적은 경우 메모리에 직접 저장됩니다 (버퍼 크기는 이 용도로 사용되는 힙 공간의 백분율을 나타내는 mapred.job.shuffle.input.buffer.percent 속성에 의해 제어됨). 데이터 양이 버퍼 크기의 일정 비율 (mapred.job.shuffle.merg.percent 에 의해 결정됨) 을 초과하면 데이터가 병합되어 디스크로 넘칩니다.

오버플로 파일이 증가함에 따라 백그라운드 스레드는 이후 병합을 위한 공간을 절약하기 위해 더 크고 질서 있는 파일로 결합합니다. 실제로 Mapreduce 는 map 과 reduce 에서 정렬 및 병합 작업을 반복합니다. 이제 누군가가 서열이 Hadoop 의 영혼이라고 말하는 이유를 이해하게 되었다.

3 병합 중에는 중간 파일 (디스크 쓰기) 이 많이 있지만 MapReduce 를 사용하면 디스크에 기록되는 데이터가 최소화되고 최종 병합 결과는 디스크에 기록되지 않고 Reduce 함수에 직접 입력됩니다.

4 감속기 입력 파일. 지속적인 통합을 통해 결국 "최종 파일" 이 생성됩니다. 왜 따옴표를 넣어야 합니까? 이 파일은 디스크나 메모리에 있을 수 있기 때문입니다. 우리에게는 Reducer 의 입력으로 메모리에 저장하고자 하지만, 기본적으로 이 파일은 디스크에 저장됩니다. Reducer 의 입력 파일이 설정되면 전체 셔플이 끝납니다. 그런 다음 Reducer 를 실행하고 결과를 HDSF 에 배치합니다.

참고: MapReduce 를 튜닝하는 것은 MapReduce Shuffle 의 성능을 튜닝하는 것입니다.

셋째, 메모리 버퍼: MapOutputBuffer

2 단계 인덱스 구조:

링 버퍼:

1 kvoffsets buffer: 위치 인덱스 kvindices 가운데 버튼/값 정보의 오프셋을 저장하는 offset index 배열이라고도 합니다. Kvoffsets 사용률이 io.sort.spill.percent (기본값 80%) 를 초과하면 SpillThread 의' 오버플로' 작업, 즉 오버플로 단계를 시작하는 작업이 트리거됩니다.

2 kv indicators buffer: 위치 인덱스 배열이라고도 하며 데이터 버퍼 kvbuffer 가운데 버튼/값의 시작 위치를 저장합니다.

3kv 버퍼 데이터 버퍼: 실제 키/값 값을 저장하는 데 사용됩니다. 기본적으로 버퍼는 io.sort.mb 의 최대 95% 를 사용할 수 있습니다. kvbuffer 사용률이 io.sort.spill.percent (기본값 80%) 를 초과하면 SpillThread 의 오버플로 작업이 트리거됩니다