並行調度演算法_Python進程之並行與並發的區別

Ⅰ 高高分！！C++，高手，MFCVC

針對你提的問題，我想，你應該還是有以下的誤區沒有解決！
對於多線程
一個基本的概念就是同時對多個任務加以控制。許多程序設計問題都要求程序能夠停下手頭的工作，改為處理其他一些問題，再返回主進程。可以通過多種途徑達到這個目的。最開始的時候，那些擁有機器低級知識的程序員編寫一些「中斷服務常式」，主進程的暫停是通過硬體級的中斷實現的。盡管這是一種有用的方法，但編出的程序很難移植，由此造成了另一類的代價高昂問題。有些時候，中斷對那些實時性很強的任務來說是很有必要的。但還存在其他許多問題，它們只要求將問題劃分進入獨立運行的程序片斷中，使整個程序能更迅速地響應用戶的請求。在一個程序中，這些獨立運行的片斷叫作「線程」（Thread），利用它編程的概念就叫作「多線程處理」。多線程處理一個常見的例子就是用戶界面。利用線程，用戶可按下一個按鈕，然後程序會立即作出響應，而不是讓用戶等待程序完成了當前任務以後才開始響應。最開始，線程只是用於分配單個處理器的處理時間的一種工具。但假如操作系統本身支持多個處理器，那麼每個線程都可分配給一個不同的處理器，真正進入「並行運算」狀態。從程序設計語言的角度看，多線程操作最有價值的特性之一就是程序員不必關心到底使用了多少個處理器。程序在邏輯意義上被分割為數個線程；假如機器本身安裝了多個處理器，那麼程序會運行得更快，毋需作出任何特殊的調校。根據前面的論述，大家可能感覺線程處理非常簡單。但必須注意一個問題：共享資源！如果有多個線程同時運行，而且它們試圖訪問相同的資源，就會遇到一個問題。舉個例子來說，兩個進程不能將信息同時發送給一台列印機。為解決這個問題，對那些可共享的資源來說（比如列印機），它們在使用期間必須進入鎖定狀態。所以一個線程可將資源鎖定，在完成了它的任務後，再解開（釋放）這個鎖，使其他線程可以接著使用同樣的資源。多線程是為了同步完成多項任務，不是為了提高運行效率，而是為了提高資源使用效率來提高系統的效率。線程是在同一時間需要完成多項任務的時候實現的。

對於STL，
從根本上說，STL是一些「容器」的集合，這些「容器」有list,vector,set,map等，STL也是演算法和其他一些組件的集合。這里的「容器」和演算法的集合指的是世界上很多聰明人很多年的傑作。STL的目的是標准化組件，這樣就不用重新開發，可以使用現成的組件。STL現在是C++的一部分，因此不用額外安裝什麽。它被內建在你的編譯器之內。

Ⅱ 並行計算模型的C3模型

C3模型假定處理機不能同時發送和接收消息，它對超步的性能分析分為兩部分：計算單元CU，依賴於本地計算量；通信單元COU，依賴與處理機發送和接收數據的多少、消息的延遲及通信引起的擁擠量。該模型考慮了兩種路由（存儲轉發路由和蟲蝕尋徑路由）和兩種發送/接收原語（阻塞和無阻塞）對COU的影響。 (1)用Cl和Cp來度量網路的擁擠對演算法性能的影響；
(2)考慮了不同路由和不同發送或接收原語對通信的影響；
(3)不需要用戶指定調度細節，就可以評估超步的時間復雜性；
(4)類似於H-PRAM模型的層次結構，C3模型給編程者提供了K級路由演算法的思路，即系統被分為K級子系統，各級子系統的操作相互獨立，用超步代替了H-PRAM中的Sub PRAM進行分割。 (1)Cl度量的前題假設為同一通信對中的2個處理機要分別位於網路對分後的不同子網路內；
(2)模型假設了網路帶寬等於處理機帶寬，這影響了正確描述可擴展系統；
(3)在K級演算法中，處理機間順序可以由多種排列，但C3模型不能區分不同排列的難易程度。

Ⅲ cpu調度的基本方式

我們知道，程序需要獲得CPU的資源才能被調度和執行，那麼當一個進程由於某種原因放棄CPU然後進入阻塞狀態，下一個獲得CPU資源去被調度執行的進程會是誰呢？下圖中，進程1因為阻塞放棄CPU資源，此時，進程2剛IO操作結束，可以獲得CPU資源去被調度，進程3的時間片輪轉結束，也同樣可以獲得CPU資源去被調度，那麼，此時的操作系統應該安排哪個進程去獲得CPU資源呢？這就涉及到我們操作系統的CPU調度策略了。

根據生活中的例子，我們很容易想到以下兩種策略CPU調度的直觀想法：1.FIFO誰先進入，先調度誰，這是一種非常簡單有效的方法，就好比我們去飯堂打飯，誰先到就給誰先打飯。但是這種策略會遇到一個問題：如果遇到一個很小的任務，但是它是最後進入的，那麼必須得前面一大堆任務結束完後才能執行這個小小的任務，這樣就感覺很不劃算呀！因為我只是簡簡單單的一個小任務，但是從打開這個任務到結束這個任務要很久。這顯然不符合我們的需求，因而我們會想到第2種策略，就是先調度小任務，後調度大任務。2.Priority很簡單，就是任務短的優先執行，但是此時又有問題了，任務雖然短，但是它的執行時間不一定短，就好比在一個銀行業務中，客戶填寫一個表，這是一個非常短的任務吧——就單單填個表，但是這個表很長很長，那麼這個短任務它的執行時間就很長了，我們怎麼知道這個短的任務將來會執行多長的時間呢？所以，這樣的策略還是依然有問題。那麼，面對諸多的場景，如何設計調度演算法呢？首先，我們要明白我們的演算法應該讓什麼更好呢？面對客戶：銀行調度演算法的設計目標應該是用戶滿意；而面對進程：CPU調度的目標應該是進程滿意。那怎麼才能讓進程滿意呢？那就是時間了。進程希望盡早地結束任務，這就是周轉時間(從任務到達到任務結束)要短，而且希望用戶的操作能夠盡快地被響應，這就是響應時間(從操作發生到響應)要短。而且系統內耗時間要少，吞吐量(任務的完成量)要大，系統需要把更多的時間用在任務的執行上，而不能老是去做無關緊要的事情，例如：頻繁切換任務，切換棧，分配資源等事情。同時，系統還要去合理地調配任務。那麼，CPU的調度策略如何做到合理呢？首先得明白系統中有以下的幾種矛盾。1.吞吐量和響應時間之間有矛盾響應時間小=>切換次數多=>系統內耗大=>吞吐量小由於需要較短的響應時間，那麼就得頻繁地切換任務，這樣系統的很多時間都花在切換任務上面了，系統的內耗大了，吞吐量就小了。2.前台任務和後台任務的關注點不同前台任務關注響應時間，後台任務關注周轉時間。前台任務例如我們的word文檔，我們打一個字，需要立馬顯示在文檔中，這就是word文檔這個任務關注的是響應時間；而後台任務中，例如我們的javac編譯java代碼，它的周轉時間要小，即該任務從進入到結束所花的時間要小，即編譯完成的時間要小。http://3.IO約束型任務和CPU約束型任務各有各的特點IO約束型任務就是使用CPU的時間較少，進行IO操作的時間較長，CPU約束型的任務就是使用CPU的時間較長。因此，要做到合理，需要折中、綜合考慮以上的幾種矛盾。由此，產生了一些CPU的調度演算法，在下一節我們將重點講述這些CPU調度演算法。

關注小鯨融創，一起深度學習金融科技！

編輯於 2019-12-11 · 著作權歸作者所有
贊同 1
評論
展開全部

Ⅳ 作業調度演算法：編寫並調度一個多道程序系統的作業調度模擬程序

回答：benben
新手
5月8日 08:33 作業調度的方法一般有：先來先服務演算法，短程作業優先演算法，響應比高者優先演算法等等把
他就是把你要處理的總的作業，根據系統允許並行的工作得道數和系統的可利用的資源，調入內存的一種演算法，如果要簡單地說就是挑選最有者的過程！

Ⅳ 最佳調度問題：假設有n個任務由k個並行工作的機器來完成。完成任務i需要的時間為Ti。試設計一個演算法找出完

Node{
int Path[n]; //節點對應的解空間樹的路徑，即到該節點為止的策略記錄
int T[k]; //在本策略下的每台機器的運行時間
int Time; //本策略的總執行時間，為每台機器運行時間的最大值
int length; //本節點的深度,即當前處理的作業
}
Proc BestDispatch(int n,int k,int t[])
Node Boot,X,P,result; //Boot為根節點，result保存最優解
int f; //記錄當前最優解的執行時間
f=n*max(t[]); //初始化f
Boot.T[n]={0};
Boot.Time=0;
Boot.Path[n]={0};
Boot.length=0; //初始化根節點
AddHeap(Boot); //根節點加入堆中，堆中元素按照Time值由小到大排序
While !Heap.empty() do
P=DeleteMinHeap(); //P為當前優先順序最高的點
for i=1 to k do //擴展P的k個子節點
X=Newnode(P.Path[],P.T[],P.length+1);
X.Path[X.length]=i;
X.T[i]=X.T[i]+t[X.length];
X.Time=max(X.T[]);
if X.length==n then //X為葉節點
if X.Time<f then //X的執行時間小於已知最優解
f=X.Time; //將X設為最優解
result=X;
end{if}
else //X為中間節點
if X.Time<f then
AddHeap(X);
end{if}//X的當前執行時間小於已知最優解則加入堆中，否則剪去
end{if}
end{for}
end{while}
end{ BestDispatch }

Ⅵ 線程的並行是真並行嗎

如果一台計算機有n個處理器，那麼就有n個線程真正同時運行。單CPU計算機是偽並行，按照某種調度演算法，多個線程輪流運行。

Ⅶ 並行處理計算機系統的結構原理

並行處理計算機的結構主要有流水線方式、多功能部件方式、陣列方式、多處理機方式和數據流方式。將指令的執行過程分解為若干段，每段進行一部分處理。一條指令順序流過所有段即執行完畢獲得結果。當本條指令在本段已被處理完畢而進入下段時，下條指令即可流入本段。因此，在整個流水線上可以同時處理若干條指令。若各段的執行時間均為一個時鍾節拍，則在正常情況下每拍可以輸出一個結果，即完成一條指令。這就可加快處理機的速度。
程序中相鄰指令的相關性會影響流水線處理機效率的發揮。例如，條件轉移指令在上條指令執行完以前，有時不能確定後繼指令；又如本條指令需要用上條指令的結果作為操作數等，都將中斷流水線而使效率下降。一台處理機由多個相同的處理部件和一個統一的控制器組成。這個控制器解釋指令並傳送操作命令至全部處理部件。各處理部件按照控制器的命令同時進行完全相同的操作。陣列處理機又可分為浮點陣列處理機和位片式陣列處理機兩類。
ILLIAC-Ⅳ機屬於浮點陣列處理機，包括64個完全相同的處理部件(PU)和一個公共的控制部件(CU)。每個處理部件包括一個能執行64位浮點操作的處理單元(PE)和一個容量為2k字的存儲器(PM)。64個處理部件排列成8×8陣列。每個處理部件與四鄰處理部件均有直接數據通路。在直接耦合多處理機系統中，實現處理機與存儲器、處理機與處理機之間連接的互連網路十分重要。互連網路有三種主要形式。
①匯流排結構：匯流排結構是多處理機系統中最為簡單的網路結構。實際的多處理機系統的互連網路，往往是在匯流排結構的基礎上發展起來的（圖3）。
②交叉開關結構：交叉開關由縱橫開關陣列組成，將橫向的處理機與縱向的存儲器模塊連接起來（圖4）。
③多埠存儲器結構：把交叉開關結構中的各交叉點上的開關移到相應存儲器的介面內部，形成多埠存儲器結構。數據流處理機是受到人們重視的高度並行的處理機。它雖保留了存儲程序的做法，但在主要原理上已與諾依曼計算機結構不同。它不按程序計數器指出的指令順序執行程序，只要所需操作數全部具備，指令即可被執行,亦即程序的執行不是由控制流驅動,而是由數據流驅動。
數據流處理機是以語言為基礎的處理機。它使用數據流程序圖作為用戶語言與計算機結構之間的介面。數據流程序圖用能動框表示。每個能動框有多個域，分別存放操作碼、操作數和目標地址。數據流程序以能動框集合的方式保存在能動存儲器中。當某條指令可以執行時，相應的能動框地址便被送入指令排隊器。讀取部件則按地址從存儲器中取出該能動框，形成操作包，送至操作部件進行處理，產生結果包。修改部件根據結果包的目標地址將結果數據送至規定的能動框作為操作數，並將具備操作數的指令的地址送至指令排隊器。指令排隊器中的指令均具備執行條件，因而只需增加部件數量或增強部件流水程度，就可以高速並行執行。此外，還可將多個指令處理單元連接成數據流多處理機系統，進一步提高處理能力。
並行演算法和並行語言
提高並行處理效率的關鍵之一是並行演算法。演算法須適應計算機的結構。如果一種演算法所表達出來的並行度與計算機的並行度基本一致，便能提高計算機的解題效率。
在向量計算機中，提高並行度的主要問題在於把可並行處理的操作數用向量表示。許多常用的數值計演算法，如數列求和、矩陣乘、高斯消元、快速傅里葉變換等，已成功地在向量計算機上實現了並行處理。較為通行的並行語言基本上是FORTRAN語言的擴展。
在多處理機系統中，提高程序並行性的關鍵，是把任務分解成足夠多的可同時操作的進程。在程序語言中，還須擴充能明確表達進程並發性的語句，以便程序運行時能為相應的控制機構提供控制和管理手段，其中包括並行任務的派生、通信和調度。ADA 語言為描述多處理機並行程序結構提供了必要的語句。為適應數據流計算機而出現的若干數據流語言如Id語言和VAL語言已經在試用。其重要特點是把數組看成是值而不是目標。用數據流語言編寫的程序能夠自然地表達出最大的運算並行性。

Ⅷ Python進程之並行與並發的區別

並行 :

當系統有一個以上CPU時，則進程的操作有可能非並發。當一個CPU執行一個進程時，另一個CPU可以執行另一個進程，兩個進程互不搶佔CPU資源，可以同時進行，這種方式我們稱之為並行。

並發 :

當有多個進程在操作時，如果系統只有一個CPU，則它根本不可能真正同時執行一個以上的進程，它只能把CPU運行時間劃分成若干個時間段，再將時間段分配給各個進程執行，在一個時間段的進程代碼運行時，其它進程處於掛起狀，這種方式我們稱之為並發。

區別：

並發和並行是即相似又有區別的兩個概念，並行是指兩個或者多個事件在同一時刻同時執行，而並發是指兩個或多個事件通過時間片輪流被執行。在多道程序環境下，並發性是指在一段時間內宏觀上有多個程序在同時運行，但在單核CPU中，同一時刻僅能有一道程序執行，故微觀上這些程序只能是分時地交替執行。倘若在計算機中有多個CPU，則這些可以並發執行的程序便可被分配到多個處理機上，實現並行執行，即利用每個處理機來處理一個可並發執行的程序，這樣，多個程序便可以同時執行。

相關推薦：

一文帶你讀懂Python中的進程

Ⅸ 在多核CPU下，同一進程下的多個線程可以並行運行嗎

CPU在某一個時間點上確實只能執行一個線程，但是多線程不是由於多核或者雙核才叫多線程。
是由於，很多個線程在並行執行的時候，CPU根據一定的線程調度演算法，頻繁的進行線程切換，當正在執行的一個線程需要進行IO操作或者需要訪問內存的時候，CPU完全可以放棄該線程，轉而調度線程就緒隊列上的其他線程，被放棄的線程則進入阻塞狀態，IO操作或者訪問內存操作結束之後，該線程可以進入線程就緒隊列上。
人們通常意義上的多線程指的是，由於CPU根據一定的線程調度演算法來切換線程，所以在一個時間段上，可以看做很多線程在並發執行。
其實還是在某一個時間點上只有一個線程在運行罷了。

Ⅹ hadoop並行過程，是由什麼機制來進行控制

可以只用一行代碼來運行MapRece作業：JobClient.runJon(conf)，Job作業運行時參與的四個實體：

1.JobClient 寫代碼，配置作業，提交作業。

2.JobTracker:初始化作業，分配作業，協調作業運行。這是一個java程序，主類是JobTracker。

3.TaskTracker：運行作業劃分後的任務，即分配數據分配上執行Map或Rece任務。

4.HDFS：保存作業數據、配置信息等，保存作業結果。

Map/Rece 作業總體執行流程：

代碼編寫 ----> 作業配置 ---->作業提交---->Map任務分配和執行---->處理中間結果----> Rece任務分配與執行----> 輸出結果

而對於每個作業的執行，又包含：

輸入准備---->任務執行---->輸出結果

作業提交JobClient：

JobClient的runJob方法產生一個Jobclient實例並調用其submitJob方法，然後runJob開始循環嗎，並在循環中調用getTaskCompetionEvents方法，獲得TaskCompletionEvent實例，每秒輪詢作業進度（後面有介紹進度和狀態更新），把進度寫到控制台，作業完成後顯示作業計數器，若失敗，則把錯誤記錄到控制台。

submitJob方法作業提交的過程：

1.向JobTracker請求一個新的JobId。

2.檢查作業相關路徑，如果路徑不正確就會返回錯誤。

3.計算作業輸入分片及其劃分信息。

4.將作業運行需要的資源（jar文件、配置文件等）復制到Shared HDFS，並

復制多個副本（參數控制，默認值為10）供tasktracker訪問，也會將計算的分片復制到HDFS。

5.調用JobTracker對象的submitJob()方法來真正提交作業，告訴JobTracker作業准備執行。

作業的初始化JobTracker：

JobTracker收到submitJob方法調用後，會把調用放入到一個內部隊列，由作業調度器（Job scheler）進行調度並對其初始化。Job初始化即創建一個作業對象。

當作業被調度後，JobTracker會創建一個代表這個作業的JobInProgress對象，並將任務和記錄信息封裝在這個對象中，以便跟蹤任務狀態和進程。

初始化過程就是JobInProgress對象的initTasks方法進行初始化的。

初始化步驟：

1.從HDFS中讀取作業對應的job.split信息，為後面的初始化做好准備。

2.創建並初始化map和rece任務。根據數據分片信息中的個數確定map task的個數，然後為每個map task生成一個TaskInProgress對象來處理數據分片，先將其放入nonRunningMapCache,以便JobTracker分配任務的時候使用。接下來根據JobConf中的mapred.rece.tasks屬性利用setNumReceTasks()方法設置rece task的數量，然後同map task創建方式。

3.最後就是創建兩個初始化task，進行map和rece的初始化。

任務的分配JobTracker：

消息傳遞HeartBeat： tasktracker運行一個簡單循環定期發送心跳（heartbeat）給JobTracker。由心跳告知JobTracker自己是否存活，同時作為消息通道傳遞其它信息（請求新task）。作為心跳的一部分，tasktracker會指明自己是否已准備好運行新的任務，如果是，jobtracker會分配它一個任務。

分配任務所屬於的作業：在Jobtracker分配任務前需先確定任務所在的作業。後面會介紹到各種作業調度演算法，默認是一個FIFO的作業調度。

分配Map和Rece任務：tasktracker有固定數量的任務槽,一個tasktracker可以同時運行多個Map和Rece任務，但其准確的數量由tasktracker的核的數量和內存大小決定。默認調度器會先填滿Map任務槽，再填Rece任務槽。jobtracker會選擇距離離分片文件最近的tasktracker，最理想情況下，任務是數據本地化（data-local）的,當然也可以是機架本地化（rack-local），如果不是本地化的，那麼他們就需要從其他機架上檢索數據。Rece任務分配很簡單，jobtracker會簡單的從待運行的rece任務列表中選取下一個來執行，不用考慮數據本地化。

任務的執行TaskTracker：

TaskTracker收到新任務後，就要在本地運行任務了，運行任務的第一步就是通過localizedJob將任務本地化所需要的注入配置、數據、程序等信息進行本地化。

1.本地化數據：從共享文件系統將job.split 、job.jar (在分布式緩存中)復制本地，將job配置信息寫入job.xml。

2.新建本地工作目錄：tasktracker會加壓job.jar文件到本工作目錄。

3.調用launchTaskForJob方法發布任務（其中會新建TaskRunner實例運行任務），如果是Map任務就啟用MapTaskRunner，對於Rece就是ReceTaskRunner。

在這之後，TaskRunner會啟用一個新的JVM來運行每個Map/Rece任務，防止程序原因而導致tasktracker崩潰，但不同任務間重用JVM還是可以的，後續會講到任務JVM重用。

對於單個Map，任務執行的簡單流程是：

1.分配任務執行參數

2.在Child臨時文件中添加map任務信息（Child是運行Map和Rece任務的主進程）

3.配置log文件夾，配置map任務的通信和輸出參數

4.讀取input split，生成RecordReader讀取數據

5.為Map生成MapRunnable,依次從RecordReader中接收數據，並調用Map函數進行處理。

6.最後將map函數的輸出調用collect收集到MapOutputBuffer（參數控制其大小）中。

Streaming和Pipes:

Streaming和Pipes都運行特殊的Map和Rece任務，目的是運行用戶提供的可執行程序並與之通信。

Streaming:使用標准輸入輸出Streaming與進程進行通信。

Pipes:用來監聽套接字，會發送一個埠號給C++程序，兩者便可建立鏈接。

進度和狀態更新：

一個作業和它的任務都有狀態（status），其中包括：運行成功失敗狀態、Map/Rece進度、作業計數器值、狀態消息。

狀態消息與客戶端的通信：

1.對於Map任務Progress的追蹤：progress是已經處理完的輸入所佔的比例。

2.對於Rece：稍復雜，rece任務分三個階段（每個階段佔1/3），復制、排序和Rece處理，若rece已執行一半的輸入的話，那麼任務進度便是1/3+1/3+1/6=5/6。

3.任務計數器：任務有一組計數器，負責對任務運行各個事件進行計數。

4.任務進度報告：如果任務報告了進度，便會設置一個標記以表明狀態將被發送到tasktracker。有一個獨立線程每隔三秒檢查一次此標記，如果已設置，則告知tasktracker當前狀態。

5.tasktracker進度報告：tasktracker會每隔5秒（這個心跳是由集群大小決定，集群越大時間會越長）發送heartbeat到jobtracker，並且tasktracker運行的所有狀態都會在調用中被發送到jobtracker。

6.jobtracker合並各任務報告：產生一個表明所有運行作業機器所含任務狀態的全局視圖。

前面提到的JobClient就是通過每秒查詢JobTracker來接收最新狀態，而且客戶端JobClient的getJob方法可以得到一個RunningJob的實例，其包含了作業的所以狀態信息。

作業的完成：

當jobtracker收到作業最後一個任務已完成的通知後，便把作業狀態設置成成功。JobClient查詢狀態時，便知道任務已成功完成，於是JobClient列印一條消息告知用戶，然後從runJob方法返回。

如果jobtracker有相應設置，也會發送一個Http作業通知給客戶端，希望收到回調指令的客戶端可以通過job.end.notification.url屬性來進行設置。

jobtracker情況作業的工作狀態，指示tasktracker也清空作業的工作狀態，如刪除中間輸出。

失敗

實際情況下，用戶的代碼存在軟體錯誤進程會崩潰，機器也會產生故障，但Hadoop能很好的應對這些故障並完成作業。

1.任務失敗

子任務異常：如Map/Rece任務中的用戶代碼拋出異常，子任務JVM進程會在退出前向父進程tasktracker發送錯誤報告，錯誤被記錄用戶日誌。tasktracker會將此次task attempt標記為tailed，並釋放這個任務槽運行另外一個任務。

子進程JVM突然退出：可能由於JVM bug導致用戶代碼造成的某些特殊原因導致JVM退出，這種情況下，tasktracker會注意到進程已經退出，並將此次嘗試標記為failed。

任務掛起：一旦tasktracker注意一段時間沒有收到進度更新，便會將任務標記為failed，JVM子進程將被自動殺死。任務失敗間隔時間通常為10分鍾，可以以作業或者集群為基礎設置過期時間，參數為mapred.task.timeout。注意：如果參數值設置為0，則掛起的任務永遠不會釋放掉它的任務槽，隨著時間的推移會降低整個集群的效率。

任務失敗嘗試次數：jobtracker得知一個tasktracker失敗後，它會重新調度該任務執行，當然，jobtracker會嘗試避免重新調度失敗過的tasktracker任務。如果一個任務嘗試次數超過4次，它將不再被重試。這個值是可以設置的，對於Map任務，參數是mapred.map.max.attempts,對於rece任務，則由mapred.rece.max.attempts屬性控制。如果次數超過限制，整個作業都會失敗。當然，有時我們不希望少數幾個任務失敗就終止運行的整個作業，因為即使有些任務失敗，作業的一些結果可能還是有用的，這種情況下，可以為作業設置在不觸發作業失敗情況下的允許任務失敗的最大百分比，Map任務和Rece任務可以獨立控制，參數為mapred.max.map.failures.percent 和mapred.max.rece.failures.percent。

任務嘗試中止（kill）：任務終止和任務失敗不同，task attempt可以中止是因為他是一個推測副本或因為它所處的tasktracker失敗，導致jobtracker將它上面的所有task attempt標記為killed。被終止的task attempt不會被計入任務運行嘗試次數，因為嘗試中止並不是任務的錯。

2.tasktracker失敗

tasktracker由於崩潰或者運行過慢而失敗，他將停止向jobtracker發送心跳（或很少發送心跳）。jobtracker注意已停止發送心跳的tasktracker（過期時間由參數mapred.tasktracker.expiry.interval設置，單位毫秒），並將它從等待調度的tasktracker池中移除。如果是未完成的作業，jobtracker會安排次tasktracker上已經運行成功的Map任務重新運行，因為此時rece任務已無法訪問（中間輸出存放在失敗的tasktracker的本地文件系統上）。

即使tasktracker沒有失敗，也有可能被jobtracker列入黑名單。如果tasktracker上面的失敗任務數量遠遠高於集群的平均失敗任務次數，他就會被列入黑名單，被列入黑名單的tasktracker可以通過重啟從jobtracker黑名單中移除。

3.jobtracker失敗

老版本的JobTracker失敗屬於單點故障，這種情況下作業註定失敗。

作業調度：

早期作業調度FIFO：按作業提交順序先進先出。可以設置優先順序，通過設置mapred.job.priority屬性或者JobClient的setJobPriority()方法制定優先順序（優先順序別：VERY_HIGH,HIGH,NORMAL,LOW,VERY_LOW）。注意FIFO調度演算法不支持搶占（preemption）,所以高優先順序作業仍然會被那些已經開始的長時間運行的低優先順序作業所阻塞。

Fair Scheler：目標是讓每個用戶公平地共享集群能力。當集群存在很多作業時，空閑的任務槽會以」讓每個用戶共享集群「的方式進行分配。默認每個用戶都有自己的作業池。FairScheler支持搶占，所以，如果一個池在特定的一段時間未得到公平地資源共享，它會終止池中得到過多的資源任務，以便把任務槽讓給資源不足的池。FairScheler是一個後續模塊，使用它需要將其jar文件放在Hadoop的類路徑下。可以通過參數map.red.jobtracker.taskScheler屬性配置（值為org.apache.hadoop.mapred.FairScheler）

Capacity Scheler：

集群由很多隊列組成，每個隊列都有一個分配能力，這一點與FairScheler類似，只不過在每個隊列內部，作業根據FIFO方式進行調度。本質上說，Capacity Scheler允許用戶或組織為每個用戶模擬一個獨立使用FIFO的集群。

shuffle和排序：

MapRece確保每個Recer的輸入都是按鍵排序的。系統執行排序的過程-將map輸出作為輸入傳給recer的過程稱為shuffle。shuffle屬於不斷被優化和改進的代碼庫的一部分，從許多方面來看，shuffle是MapRece的心臟。

整個shuffle的流程應該是這樣：

map結果劃分partition 排序sort 分割spill 合並同一劃分合並同一劃分合並結果排序 rece處理輸出

Map端：

寫入緩沖區：Map函數的輸出，是由collector處理的，它並不是簡單的將結果寫到磁碟。它利用緩沖的方式寫到內存，並處於效率的考慮進行預排序。每個map都有一個環形的內存緩沖區，用於任務輸出，默認緩沖區大小為100MB（由參數io.sort.mb調整），一旦緩沖區內容達到閾值（默認0.8），後台進程邊開始把內容寫到磁碟（spill），在寫磁碟過程中，map輸出繼續被寫到緩沖區，但如果緩沖區被填滿，map會阻塞知道寫磁碟過程完成。寫磁碟將按照輪詢方式寫到mapred.local.dir屬性制定的作業特定子目錄中。

寫出緩沖區：collect將緩沖區的內容寫出時，會調用sortAndSpill函數，這個函數作用主要是創建spill文件，按照key值對數據進行排序，按照劃分將數據寫入文件，如果配置了combiner類，會先調用combineAndSpill函數再寫文件。sortAndSpill每被調用一次，就會寫一個spill文件。

合並所有Map的spill文件：TaskTracker會在每個map任務結束後對所有map產生的spill文件進行merge，merge規則是根據分區將各個spill文件中數據同一分區中的數據合並在一起，並寫入到一個已分區且排序的map輸出文件中。待唯一的已分區且已排序的map輸出文件寫入最後一條記錄後，map端的shuffle階段就結束了。

在寫磁碟前，線程首先根據數據最終要傳遞到的recer把數據劃分成響應的分區（partition），在每個分區中，後台線程按鍵進行內排序，如果有一個combiner,它會在排序後的輸出上運行。

內存達到溢出寫的閾值時，就會新建一個溢出寫文件，因為map任務完成其最後一個輸出記錄之後，會有幾個溢出寫文件。在任務完成前，溢出寫文件會被合並成一個已分區且已排序的輸出文件。配置屬性io.sort.facor控制一次最多能合並多少流，默認值是10。

如果已經指定combiner,並且寫次數至少為3（通過min.mum.spills.for.combine設置）時，則combiner就會在輸出文件寫到磁碟之前運行。運行combiner的意義在於使map輸出更緊湊，捨得寫到本地磁碟和傳給recer的數據更少。

寫磁碟時壓縮：寫磁碟時壓縮會讓寫的速度更快，節約磁碟空間，並且減少傳給recer的數據量。默認情況下，輸出是不壓縮的，但可以通過設置mapred.compress.map.output值為true，就可以啟用壓縮。使用的壓縮庫是由mapred.map.output.compression.codec制定。

recer獲得文件分區的工作線程：recer通過http方式得到輸出文件的分區，用於文件分區的工作線程數量由tracker.http.threads屬性指定，此設置針對的是每個tasktracker，而不是每個map任務槽。默認值為40，在大型集群上此值可以根據需要而增加。

Rece端：

復制階段：rece會定期向JobTracker獲取map的輸出位置，一旦拿到輸出位置，rece就會從對應的TaskTracker上復制map輸出到本地（如果map輸出很小，則會被復制到TaskTracker節點的內存中，否則會被讓如磁碟），而不會等到所有map任務結束（當然這個也有參數控制）。

合並階段：從各個TaskTracker上復制的map輸出文件（無論在磁碟還是內存）進行整合，並維持數據原來的順序。

Rece階段：從合並的文件中順序拿出一條數據進行rece函數處理，然後將結果輸出到本地HDFS。

Map的輸出文件位於運行map任務的tasktracker的本地磁碟，現在，tasktracker要為分區文件運行rece任務。每個任務完成時間可能不同，但是只要有一個任務完成，rece任務就開始復制其輸出，這就是rece任務的復制階段（ phase）。rece任務有少量復制線程，因此能夠並行取得map輸出。默認值是5個線程，可以通過mapred.rece.parallel.copies屬性設置。

Recer如何得知從哪個tasktracker獲得map輸出：map任務完成後會通知其父tasktracker狀態已更新，tasktracker進而通知（通過heart beat）jobtracker。因此，JobTracker就知道map輸出和tasktracker之間的映射關系，recer中的一個線程定期詢問jobtracker以便獲知map輸出位置。由於recer有可能失敗，因此tasktracker並沒有在第一個recer檢索到map輸出時就立即從磁碟上刪除它們，相反他會等待jobtracker告示它可以刪除map輸出時才刪除，這是作業完成後最後執行的。

如果map輸出很小，則會被直接復制到rece tasktracker的內存緩沖區（大小由mapred.job.shuffle.input.buffer.percent控制，占堆空間的百分比），否則，map輸出被復制到磁碟。一旦內存緩沖區達到閾值大小（由mapred.iob.shuffle.merge.percent）

或達到map輸出閾值大小（mapred.inmem.threadhold），則合並後溢出寫到磁碟中。

隨著磁碟上副本增多，後台線程會將他們合並為更大的、排好序的文件。注意：為了合並，壓縮的map輸出必須在內存中被解壓縮。

排序階段：復制階段完成後，rece任務會進入排序階段，更確切的說是合並階段，這個階段將合並map輸出，維持其順序排列。合並是循環進行的，由合並因子決定每次合並的輸出文件數量。但讓有可能會產生中間文件。

rece階段：在最後rece階段，會直接把排序好的文件輸入rece函數，不會對中間文件進行再合並，最後的合並即可來自內存，也可來自磁碟。此階段的輸出會直接寫到文件系統，一般為hdfs。

細節：這里合並是並非平均合並，比如有40個文件，合並因子為10，我們並不是每趟合並10個，合並四趟。而是第一趟合並4個，後三趟合並10，在最後一趟中4個已合並的文件和餘下6個未合並會直接並入rece。

導航:首頁 > 源碼編譯 > 並行調度演算法

並行調度演算法

與並行調度演算法相關的資料