一文了解 Hadoop 運行機制

時間 2019-11-11

原文原文鏈接

大數據技術棧在當下已是比較成熟的了，Hadoop 做爲大數據存儲的基石，其重要程度不言而喻，做爲一個想從 java 後端轉向大數據開發的程序員來講，打好 Hadoop 基礎，就至關於夯實建造房屋的地基，本文以上圖結構爲基本，旨在幫助你們快速瞭解 Hadoop 運行機制。css

HDFS 篇java

HDFS就是你們熟知的分佈式存儲的文件系統，它包括 3 個組件，結構以下圖：node

NameNode 至關於 Master 節點，它是管理者；nginx
DataNode 是 Slave，是執行實際操做的節點；程序員
SecondryNameNode並非 NameNode的熱備，它是輔助 NameNode工做的。算法

下面咱們來從圖示的幾個方面來進行詳細闡述，如圖：數據庫

HDFS 寫數據流程編程

客戶端向 NameNode 發送上傳文件請求，附帶上傳路徑；後端
NameNode 檢查路徑是否存在若是已經存在則報出「路徑已存在」錯誤，反正通知客戶端能夠上傳文件；緩存
客戶端將文件根據拆分後，請求上傳第一個塊（0-128M），向NameNode 查找可上傳的 DataNode 信息；
NameNode 返回可用的 DataNode 信息，按副本個數返回可用DataNode 個數；
客戶端收到 DataNode 信息後，請求與其中一個 DataNode 創建 block 傳輸通道，創建成功後，第二個 DataNode 與此 DataNode 創建 block 通道，第三個 DataNode 與第二個 DataNode 創建 block 通道，以此類推，DataNode 之間逐級創建通道完成
通道創建完成後，從最後一個 DataNode 向倒數第二個 DataNode 應答通道創建成功，倒數第二個再向它前一個DataNode 應答通道創建成功，直到第一個 DataNode 應答客戶端通道創建成功；
客戶端開始傳輸數據，將block 拆成一個個的 package 傳輸，當傳到第一個 DataNode 時，將 package 放到內存中，邊將package 存儲，邊將內存中的數據傳輸到下一個 DataNode，以此方法傳到最後一個DataNode。

這樣作的好處是：加快傳輸速度並減小其餘DataNode等待時間提升傳輸效率。
當第一個塊傳輸完成時，再請求傳輸第二個block 重複 3-8 步。

HDFS 讀取數據流程

客戶端向 NameNode 請求下載 xxx 路徑下的文件；
NameNode 返回目標文件的元數據信息，即文件在 DataNode 上存儲位置，因爲一個文件是拆分存儲在不一樣的DataNode上，因此客戶端須要與不一樣DataNode 進行交互；
客戶端請求讀取第一個 DataNode數據；
第一個 DataNode 數據向客戶端傳輸數據 block_1
第一個DataNode 數據讀取完成後，向第二個 DataNode請求讀取數據，第二個DataNode 想客戶端傳輸文件 block_2
重複以上步驟，直到讀取文件完成

NameNode 與 SecondryNameNode 工做機制

Hadoop在第一次啓動時須要格式化 NameNode，NameNode在格式化後會建立 fsimage和 edits ，若是不是第一次啓動則直接加載編輯日誌和鏡像文件；
客戶端對元數據進行增刪改的請求；
NameNode 記錄操做日誌，更新滾動日誌；
NameNode 在內存中對數據進行增刪改查；
SecondaryNameNode 詢問 NameNode 是否須要checkpoint，直接帶回 NameNode 是否檢查結果；
SecondaryNameNode 請求執行 checkpoint；
NameNode 滾動正在寫的 edits 日誌；
將滾動前的編輯日誌和鏡像文件拷貝到 SecondaryNameNode；
SecondaryNameNode 加載編輯日誌和鏡像文件到內存，併合並；
生成新的鏡像文件 fsimage.chkpoint；
拷貝f simage.chkpoint 到 NameNode；
NameNode 將 fsimage.chkpoint 從新命名成 fsimage。

fsimage 與 edits 以及其餘文件介紹：

fsimage文件：HDFS 文件系統元數據的一個永久性的檢查點，其中包含HDFS文件系統的全部目錄和文件 idnode 的序列化信息。

edits文件：存放HDFS文件系統的全部更新操做的路徑，文件系統客戶端執行的全部寫操做首先會被記錄到edits文件中。

seen_txid文件保存的是一個數字，就是最後一個edits_的數字

每次 NameNod e啓動的時候都會將 fsimage 文件讀入內存，並將 edits裏面的更新操做，保證內存中的元數據信息是最新的、同步的，能夠當作 NameNode 啓動的時候就將 fsimage 和 edits 文件進行了合併。

checkpoint 檢查點設置

SecondaryNameNode 每隔一小時執行一次檢查，當操做次數達到1百萬時，執行一次檢查。具體配置在 hdfs-default.xml 中。

DataNode 工做機制

DataNode 啓動後向 NameNode 註冊，經過後，週期性（1小時）的向NameNode 上報全部的塊信息。
心跳是每 3 秒一次，心跳返回結果帶有 NameNode 給該 DataNode 的命令如複製塊數據到另外一臺機器，或刪除某個數據塊。若是超過 10 分鐘沒有收到某個 DataNode 的心跳，則認爲該節點不可用，不會立刻斷定爲該節點死亡。

HDFS 默認的超時時長爲 10 分鐘+30 秒。若是定義超時時間爲 timeout，則超時時長的計算公式爲：

 timeout  = 2 * dfs.namenode.heartbeat.recheck-interval + 10 * dfs.heartbeat.interval

而默認的 dfs.namenode.heartbeat.recheck-interval 大小爲 5 分鐘，dfs.heartbeat.interval 默認爲 3 秒。

安全模式

NameNode 啓動時，首先將映像文件（fsimage）載入內存，並執行編輯日誌（edits）中的各項操做，這是須要必定時間的，此時 NameNode 運行在安全模式，即 NameNode 的文件系統對於客戶端來講是隻讀的。

系統中的數據塊的位置並非由 NameNode 維護的，而是以塊列表的形式存儲在 DataNode 中。

在系統的正常操做期間，NameNode 會在內存中保留全部塊位置的映射信息。在安全模式下，各個 DataNode 會向 NameNode 發送最新的塊列表信息，NameNode 瞭解到足夠多的塊位置信息以後，便可高效運行文件系統。

MapReduce篇

mapreduce 是一種並行計算的框架，它是 hadoop 並行計算的基礎，下面咱們從圖示的幾個方面來開始 MR 的介紹

MR 思想

分佈式的運算程序每每須要分紅至少 2 個階段。

第一個階段的 maptask 併發實例，徹底並行運行，互不相干。

第二個階段的 reduce task 併發實例互不相干，可是他們的數據依賴於上一個階段的全部 maptask 併發實例的輸出。

MapReduce 編程模型只能包含一個 map 階段和一個 reduce 階段，若是用戶的業務邏輯很是複雜，那就只能多個 mapreduce 程序，串行運行。

MR 編程規範與大致流程

Mapper 階段

用戶自定義的 Mapper 要繼承本身的父類
Mapper 的輸入數據是 KV 對的形式（KV 的類型可自定義）
Mapper 中的業務邏輯寫在 map() 方法中
Mapper 的輸出數據是 KV 對的形式（KV 的類型可自定義）
map() 方法（maptask 進程）對每個<K,V>調用一次

Reducer 階段

用戶自定義的 Reducer 要繼承本身的父類
Reducer 的輸入數據類型對應 Mapper 的輸出數據類型，也是 KV
Reducer 的業務邏輯寫在 reduce() 方法中
Reducetask 進程對每一組相同 k 的<k,v>組調用一次 reduce() 方法

Driver階段

至關於 yarn 集羣的客戶端，用於提交咱們整個程序到 yarn 集羣，提交的是封裝了 mapreduce 程序相關運行參數的 job 對象

分片

分片機制：

簡單地按照文件的內容長度進行切片
切片時不考慮數據集總體，而是逐個針對每個文件單獨切片
切片大小，默認等於 block 大小，
切片判斷，當剩下的文件大小大於切片大小的 1.1 倍時才進行切片

切片大小： Math.max(minSize, Math.min(maxSize,blockSize));

切片主要由這幾個值來運算決定：

mapreduce.input.fileinputformat.split.minsize = 1 默認值爲 1

mapreduce.input.fileinputformat.split.maxsize = Long.MAXValue 默認值 Long.MAXValue

所以，默認狀況下，切片大小 = blocksize。

輸入流 FileInputFormat 與自定義：

經常使用的輸入流 TextFileInputFormat（按行讀入），combinerTextInputFormat（讀入時合併小分件）等

自定義輸入流須要繼承 FileInputFormat ，重寫 createRecordReader 方法。

MapTask機制

MapTask 的並行度是由切片的個數決定的，有多少個切片就會執行多少個 MapTask。大體流程以下圖：

Read階段：MapTask 經過用戶編寫的 RecordReader，從輸入切片中解析出一個個 key/value。
Map 階段：該節點主要是將解析出的 key/value 交給用戶編寫map()函數處理，併產生一系列新的 key/value。
Collect收集階段：在用戶編寫 map( )函數中，當數據處理完成後，通常會調用 OutputCollector.collect() 輸出結果。在該函數內部，它會將生成的 key/value 分區（調用 Partitioner），並寫入一個環形內存緩衝區中。
Spill階段：即「溢寫」，當環形緩衝區滿後，MapReduce 會將數據寫到本地磁盤上，生成一個臨時文件。須要注意的是，將數據寫入本地磁盤以前，先要對數據進行一次本地排序，並在必要時對數據進行合併、壓縮等操做。

溢寫階段詳情：

步驟1：利用快速排序算法對緩存區內的數據進行排序，排序方式是，先按照分區編號 partition 進行排序，而後按照 key 進行排序。這樣，通過排序後，數據以分區爲單位彙集在一塊兒，且同一分區內全部數據按照 key 有序。
步驟2：按照分區編號由小到大依次將每一個分區中的數據寫入任務工做目錄下的臨時文件 output/spillN.out（N 表示當前溢寫次數）中。若是用戶設置了 Combiner，則寫入文件以前，對每一個分區中的數據進行一次彙集操做。
步驟3：將分區數據的元信息寫到內存索引數據結構 SpillRecord中，其中每一個分區的元信息包括在臨時文件中的偏移量、壓縮前數據大小和壓縮後數據大小。若是當前內存索引大小超過 1MB，則將內存索引寫到文件 output/spillN.out.index 中。

5 .Combine階段：當全部數據處理完成後，MapTask對全部臨時文件進行一次合併，以確保最終只會生成一個數據文件。

在進行文件合併過程當中，MapTask 以分區爲單位進行合併。對於某個分區，它將採用多輪遞歸合併的方式。每輪合併 io.sort.factor（默認100）個文件，並將產生的文件從新加入待合併列表中，對文件排序後，重複以上過程，直到最終獲得一個大文件。

讓每一個 MapTask 最終只生成一個數據文件，可避免同時打開大量文件和同時讀取大量小文件產生的隨機讀取帶來的開銷。

Shuffle 機制

Mapreduce 確保每一個 reducer 的輸入都是按 key 排序的。系統執行排序的過程（即將 mapper 輸出做爲輸入傳給 reducer）稱爲 shuffle，如圖下圖所示。

shuffle 在map 階段主要是進行分區、排序、合併，在reduce 階段主要進行合併、排序。

Partition 分區

默認分區是根據 key 的 hashCode 和 Integer的最大值作與運算後對reduceTasks個數取模獲得的。

return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;

使用默認規則咱們無法控制哪一個key存儲到哪一個分區因此一般須要自定義分區規則，自定義分區的步驟很簡單：

自定義類繼承 Partitioner，重寫 getPartition() 方法
在 job 驅動中，設置自定義partitioner，job.setPartitionClass(自定義類)
自定義 partition 後，要根據自定義 partitioner 的邏輯設置相應數量的 reduce task

此時要注意分區個數與 reduceTask 個數的關係：

若是 reduceTask 的數量> 分區數，則會多產生幾個空的輸出文件；

若是 1< reduceTask 的數量< 分區數，則有一部分分區數據無處安放，程序會報錯；

若是 reduceTask 的數量 =1，則無論 mapTask 端輸出多少個分區文件，最終結果都交給這一個 reduceTask，最終也就只會產生一個結果文件，實際上這種狀況是不會進行分區的。

排序

shuffle 過程當中按排序時機來看一共進行了4次排序：

緩衝區在寫入磁盤前按天然序先對文件進行快速排序
在 combiner 階段對文件進行合併排序，歸併排序法
reduceTask 將 MapTask 的輸出文件拷貝後進行歸併排序
進入 reduce 方法前進行自定義排序，使用 GroupingComparato r組件

GroupingComparator 經常使用的例子是自定義訂單 comparator，根據訂單id排序，大體步驟以下：

建立訂單對象繼承 WritableComparator ,重寫 compare方法返回1,-1,0
自定義GroupingComparator對象繼承WritableComparator,重寫compare方法，傳入訂單對象作比較
job.setGroupingComparatorClass（自定義GroupingComparator）
若有須要再根據訂單id，自定義分區

合併

combiner 是 MR 程序中 Mapper 和 Reducer 以外的一種組件，它主要用場景是在溢寫階段在寫入到磁盤前對文件先進行一次合併操做，須要注意的是應用 combiner 的前提是不能影響最終的業務邏輯

其實 combiner 組件的父類就是 Reducer，combiner 和 reducer 的區別在於運行的位置：

Combiner是在每個 maptask 所在的節點運行;它的意義就是對每個 maptask 的輸出進行局部彙總，以減少網絡傳輸量。

Reducer 是接收全局全部 Mapper 的輸出結果；

combiner 的使用：

自定義 combiner 類繼承 reducer ，重寫 reduce 方法
在 Driver 中 job.setCombinerClass（自定義 combiner 類）

ReduceTask機制

reducetask的並行度一樣影響整個 job 的執行併發度和執行效率，但與 maptask 的併發數由切片數決定不一樣，Reducetask 數量的決定是能夠直接手動設置：

//默認值是1，手動設置爲 5job.setNumReduceTasks(5);

reduce 個數若是是 0，則沒有 reduce 過程輸出文件數和 map 數一致

reduce 個數若是是 1，則輸出一個文件，不會有分區過程

大體流程以下圖所示：

Copy 階段：ReduceTask 從各個 MapTask 上遠程拷貝一片數據，並針對某一片數據，若是其大小超過必定閾值，則寫到磁盤上，不然直接放到內存中。
Merge 階段：在遠程拷貝數據的同時，ReduceTask 啓動了兩個後臺線程對內存和磁盤上的文件進行合併，以防止內存使用過多或磁盤上文件過多。
Sort 階段：按照 MapReduce 語義，用戶編寫 reduce() 函數輸入數據是按 key 進行彙集的一組數據。爲了將 key 相同的數據聚在一塊兒，Hadoop 採用了基於排序的策略。因爲各個 MapTask 已經實現對本身的處理結果進行了局部排序，所以，ReduceTask 只需對全部數據進行一次歸併排序便可。
Reduce 階段：reduce() 函數將計算結果寫到 HDFS 上。（默認是HDFS，具體寫到哪要看輸出流）

輸出流 FileOutputFormat 與自定義：

經常使用的輸入接口 TextOutputFormat（每條記錄寫成行，字符串輸出），SequenceFileOutputFormat（順序文件輸出須要做爲後序MR輸入）等。

自定義輸出流須要繼承FileOutputFormat ，重寫 getRecordWriter方法，返回自定義RecordWriter。

join

提到 join 最多的是使用 map join，緣由在於使用 reduce join 會形成 reduce 端壓力大，容易產生數據傾斜，

使用map join的兩種狀況：

一張小表一張大表狀況，小表驅動大表

在map端緩存多張表，提早處理業務邏輯，在map端多處理業務，減小reduce端壓力，儘量減小數據傾斜

MR 總結

因爲 MR 部份內容過多而且複雜，咱們用一張圖來簡單總結下核心知識點，具體描述在上文均有提到：

壓縮

對於 Hadoop 來講壓縮特性運用得當能提升性能，但運用不當也可能下降性能。

有以下兩個基本原則：

運算密集型的job，少用壓縮
IO密集型的job，多用壓縮

YARN

YARN 架構組成

YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等組件構成，

ResourceManager

處理客戶端請求
監控 NodeManager
啓動或監控 ApplicationMaster
資源的分配與調度

NodeManager

管理節點資源
處理 ResourceManager 請求
處理 ApplicationMaster 請求

ApplicationMaster

數據切分
爲應用程序申請資源並分配給內部的任務
任務監控與容錯

Container

Container是 YARN 中資源抽象，封裝了某個節點上的資源內存 CPU 磁盤網絡

YARN做業機制

YARN 的做業機制是比較宏觀的，它反映了一個 MR 任務從提交到集羣到完成的整個週期，如圖：

做業提交全過程詳解

做業提交

1.client 調用 job.waitForCompletion 方法，向整個集羣提交MapReduce 做業;

2.client 向 RM 申請一個做業 id;

3.RM 給 client 返回該 job 資源的提交路徑和做業 id;

4.client 提交 jar 包、切片信息和配置文件到指定的資源提交路徑;

5.client 提交完資源後，向 RM 申請運行 MrAppMaster;
做業初始化

6.當 RM 收到 client 的請求後，將該 job 添加到容量調度器中;

7.某一個空閒的 NM 領取到該 job;

8.該 NM 建立 Container，併產生 MRAppmaster;

9.下載 client 提交的資源到本地;

任務分配

10.MrAppMaster 向 RM 申請運行多個 maptask 任務資源;

11.RM 將運行 maptask 任務分配給另外兩個 NodeManager，另兩個 NodeManager 分別領取任務並建立容器;

任務運行

12.MR 向兩個接收到任務的 NodeManager 發送程序啓動腳本，這兩個 NodeManager 分別啓動 maptask，maptask 對數據分區排序;

13.MrAppMaster 等待全部 maptask 運行完畢後，向 RM 申請容器，運行 reducetask ;

14.reduce task 向 maptask 獲取相應分區的數據;

15.程序運行完畢後，MR 會向 RM 申請註銷本身;

進度和狀態更新

YARN 中的任務將其進度和狀態返回給應用管理器, 客戶端每秒嚮應用管理器請求進度更新, 展現給用戶。

做業完成

除了嚮應用管理器請求做業進度外, 客戶端每 5 分鐘都會經過調用 waitForCompletion() 來檢查做業是否完成。

做業完成以後, 應用管理器和 container 會清理工做狀態。做業的信息會被做業歷史服務器存儲以備以後用戶覈查。

調度機制

hadoop 資源調度器分爲隊列 FIFO（先進先出調度器）、容量調度器、公平調度器，默認使用的是容量調度器。

容量調度器

組成：

多個隊列組成，每一個隊列可配置必定資源，隊列採用FIFO

對同一用戶提交做業所需資源進行限定

算法：

1. 縱向排序- 先計算每一個隊列中正在運行的任務數與其應該分得的計算資源的比值，按比值從小到大給隊列排序

2. 橫向排序- 按照做業優先級和提交順序，同時考慮用戶資源限制和內存限制對該隊列內任務排序

3. 不一樣隊列中排在第一位的任務同時執行

4. 隊列中排在第一位的任務優先執行，其餘任務依次執行

公平調度器

按缺額排序，缺額大者優先 ,缺額是指 job 在理想狀況下須要的計算資源與實際獲得的計算資源的差值

組成:

多個隊列組成，每一個隊列中每一個做業公平共享隊列資源

算法:

1.同一個隊列中，job的缺額越大，越先得到的資源，優先運行

2.做業按缺額的高低來前後執行：

a.同一隊列中可能有多個任務同時執行

b.不一樣對列中可能有多個任務同時執行

企業優化

最後咱們來談談hadoop在企業中經常使用的優化措施：

輸入端小文件處理

1. 合併小文件,對小文件進行存檔，將多個小文件打包成一個 HAR 文件，命令以下：

hadoop archive -archiveName zoo.har -p / foo / bar （打包路徑） -r 3 （副本數）/ outputdir（打包目錄）

2. 自定義 InputFormat 將小文件存儲成 SequenceFile

3. 使用 CombineTextInputFormat 來是做爲輸入，解決輸入端大量小文件場景,

4. 對於大量小文件 Job，能夠開啓 JVM 重用

開啓前一個 map 運行一個 jvm，開啓後在一個 map 在 jvm 上運行完畢後，jvm 繼續運行其餘 map。

具體設置：mapreduce.job.jvm.numtasks值在10-20之間。

Map階段

1. 增大環形緩衝區大小，由100M擴大到200M

mapreduce.task.io.sort.mb(shuffle的環形緩衝區大小，默認100m)

2. 增大緩衝區溢寫比例，由80%擴大到90%

mapreduce.map.sort.spill.percent (環形緩衝區溢出的閾值，默認80%)

3. 減小溢寫文件的 merge 次數

4. 不影響實際業務的前提下，使用 combiner 進行合併，減小I/O

reduce階段

1. 合理設置map和reduce數，不能太多特不能太少。太少會致使task等待，延長處理時間；太多致使map reduce任務間競爭，形成處理處理

2. 規避使用 reduce 由於reduce在用於鏈接數據集時會產生大量的網絡消耗

3. 設置 map reduce共存，調整slowstart.completedmaps 參數，使map 運行到必定程度後，reduce 也開始運行，減小reduce 的等待時間

4 增長每一個reduce去map中拿數據的並行數

5 集羣性能能夠的前提下，增大reduce端存儲數據內存的大小

IO傳輸

1. 採用數據壓縮的方式，減小網絡IO的的時間。安裝Snappy和LZOP壓縮編碼器,

2. 使用SequenceFile二進制文件

總體優化，經常使用調優參數表

1. MapTask默認內存大小爲1G，能夠增長MapTask內存大小爲4-5G

2. ReduceTask默認內存大小爲1G，能夠增長ReduceTask內存大小爲4-5G

3. 能夠增長MapTask的cpu核數，增長ReduceTask的cpu核數

4. 增長每一個container的cpu核數和內存大小

5. 調整每一個Map Task和Reduce Task最大重試次數

具體參數參考以下表：

資源相關參數

如下參數是在用戶本身的mr應用程序中配置就能夠生效（mapred-default.xml）

配置參數	參數說明
mapreduce.map.memory.mb	一個Map Task可以使用的資源上限（單位:MB），默認爲1024。若是Map Task實際使用的資源量超過該值，則會被強制殺死。
mapreduce.reduce.memory.mb	一個Reduce Task可以使用的資源上限（單位:MB），默認爲1024。若是Reduce Task實際使用的資源量超過該值，則會被強制殺死。
mapreduce.map.cpu.vcores	每一個Map task可以使用的最多cpu core數目，默認值: 1
mapreduce.reduce.cpu.vcores	每一個Reduce task可以使用的最多cpu core數目，默認值: 1
mapreduce.reduce.shuffle.parallelcopies	每一個reduce去map中拿數據的並行數。默認值是5
mapreduce.reduce.shuffle.merge.percent	buffer中的數據達到多少比例開始寫入磁盤。默認值0.66
mapreduce.reduce.shuffle.input.buffer.percent	buffer大小佔reduce可用內存的比例。默認值0.7
mapreduce.reduce.input.buffer.percent	指定多少比例的內存用來存放buffer中的數據，默認值是0.0

如下參數應該在yarn啓動以前就配置在服務器的配置文件中才能生效（yarn-default.xml）

配置參數	參數說明
yarn.scheduler.minimum-allocation-mb	給應用程序container分配的最小內存，默認值：1024
yarn.scheduler.maximum-allocation-mb	給應用程序container分配的最大內存，默認值：8192
yarn.scheduler.minimum-allocation-vcores	每一個container申請的最小CPU核數，默認值：1
yarn.scheduler.maximum-allocation-vcores	每一個container申請的最大CPU核數，默認值：32
yarn.nodemanager.resource.memory-mb	給containers分配的最大物理內存，默認值：8192

如下參數是 shuffle 性能優化的關鍵參數，應在yarn啓動以前就配置好（mapred-default.xml）

配置參數	參數說明
mapreduce.task.io.sort.mb	shuffle的環形緩衝區大小，默認100m
mapreduce.map.sort.spill.percent	環形緩衝區溢出的閾值，默認80%

容錯相關參數(mapreduce性能優化)

配置參數	參數說明
mapreduce.map.maxattempts	每一個Map Task最大重試次數，一旦重試參數超過該值，則認爲Map Task運行失敗，默認值：4。
mapreduce.reduce.maxattempts	每一個Reduce Task最大重試次數，一旦重試參數超過該值，則認爲Map Task運行失敗，默認值：4。
mapreduce.task.timeout	Task超時時間，常常須要設置的一個參數，該參數表達的意思爲：若是一個task在必定時間內沒有任何進入，即不會讀取新的數據，也沒有輸出數據，則認爲該task處於block狀態，多是卡住了，也許永遠會卡住，爲了防止由於用戶程序永遠block住不退出，則強制設置了一個該超時時間（單位毫秒），默認是600000。若是你的程序對每條輸入數據的處理時間過長（好比會訪問數據庫，經過網絡拉取數據等），建議將該參數調大，該參數太小常出現的錯誤提示是「AttemptID:attempt_14267829456721_123456_m_000224_0 Timed out after 300 secsContainer killed by the ApplicationMaster.」。