李亞坤：Hadoop YARN在位元組跳動的實踐

【IT168 專稿】本文根據李亞坤老師在2018年10月17日【第十屆中國系統架構師大會】現場演講內容整理而成。

講師簡介：

李亞坤，哈工大碩士，目前從事分散式計算資源調度系統YARN的研發支持工作，支撐了包括今日頭條、抖音短視頻、火山小視頻、西瓜視頻等一系列產品的離線、流式計算任務。

摘要：

位元組跳動公司的今日頭條、抖音短視頻、火山小視頻、西瓜視頻等一系列產品，在最近幾年內數據量一直呈現出爆炸性增長趨勢，數據基礎架構部門在離線計算、流式計算等多個方向上遭遇到了一系列前所未有的挑戰。本次演講主要介紹從0到4萬+計算結點的YARN集群管理，以及在調度優化、流式作業支持等多個方面的經驗分享。

分享大綱：

1、IntroductiontoYARN

2、YARN@ByteDance Overview

3、Customization@ByteDance

4、Future Works

正文：

1、IntroductiontoYARN

首先，我們將Hadoop生態圈分為三層，從底向上依次是物理層、分散式系統層和用戶層。物理層由一系列標準X86伺服器組成，這些伺服器上跑著開源Linux或者Windows系統。中間是分散式系統層，分別由HDFS服務管理的分散式存儲、Yarn管理的分散式計算資源，以及其上的一些計算框架，比如MapReduce、Storm和Flink等組成，這些都是為了方便用戶使用並編寫分散式應用。最上面是應用層，主要存放日報、進行數據分析，還有推薦模型等訓練。本文將重點介紹Yarn所在的分散式計算資源管理層。

上圖為Yarn架構圖，Yarn中有兩大概念需要普及：一是Resource Manager，這是Yarn的調度大腦。二是NodeManager，由很多Manager組成，是最基本的Yarn計算節點服務，負責執行並監控用戶作業。無論是Spark、MapReduce還是Flink，都必須按照Yarn對作業的抽象進行編程，這樣才能跑在Yarn之上。

2、YARN@ByteDance Overview

接下來，我將從不同視角向大家介紹Yarn在位元組跳動的概覽情況。從物理資源來看，所有計算節點全部由基礎架構團隊管理，目前的流式作業資源佔比不到30%，但增長速度很快。今年9月份與去年3月份對比，集群總規模大概漲了15倍。從集群視角出發，我們現在有400左右隊列，100左右Label，十個集群大概分布在五個數據中心。從作業視角出發，我們每天大概要完成34萬作業，其中，MapReduce大概30萬，Spark大概4萬，Streaming作業大概2000個。從用戶視角出發，我們目前日活用戶2000左右(公司內部)，月活用戶3000左右(公司內部)。我們的用戶可從兩個維度看，從橫向業務維度看，所有對外APP基本都在使用該服務，也有部分公司內部自研且僅為內部提供的基礎工具軟體。如果從職責上看，上層的推薦、廣告和分析，下層的系統運維包括機房網路管理等，都在使用Yarn服務做運算。

3、Customization@ByteDance

接下來，我將介紹定製化方面的事情。首先，我們在使用的Yarn基於社區2.6.0源碼，我們大概用了三年多時間，這之中也發現了很多問題，比如在單機群5000台機器時，一個簡單的切主就可能導致集群掛掉，比如Yarn原生搶佔機制在該版本下有bug。在使用過程中，由於我們的規模較大，因此對穩定性要求更高。同時，由於用戶較多，我們在易用性上也做了改進。

首先，我們來看在調度層面做的定製化。為了提高節點資源利用率而做的優化叫Delay scheduling for Dominant Resource Fairness，這是一個延遲調度。為了主資源的公平性，延遲調度與社區提到的Delay scheduling是完全不同的概念，社區提到的Delay scheduling是根據node的locality進行Delay，如果分配不滿足locality就要等待，而我們則是為了節點資源等待。一台機器的計算資源其實有很多維度，CPU和內存是最常見的兩個，還有網路IO、磁碟IO和GPU等。以CPU和內存為例，假設現在有一個container，需要申請兩個CPU和1G內存，因為Yarn是一個心跳分配機制，如果在11:01時來了一個節點node1，現在閑置了兩個CPU和8G內存，理論上是可以分配的，因為我只需要兩個CPU和1G內存。如果分配，這台node1的機器就會完全用完CPU，而內存還剩下87%，這部分內存基本就不會有人使用了，這其實是一個非常嚴重的資源碎片。如果跳過node1，在一段時間之後，node2來了，node2需要1個CPU，2G內存，這也是可以匹配的，並且將資源分配到node2之後，CPU和內存都會有一些剩餘，而這些剩餘其實可以被用來調度其他作業，資源利用就會得到很大提升，碎片化會降低很多。這套機制也讓我們的集群在高峰時段的資源碎片化小於10%，基本維持在5%左右。

但是，這類優化會帶來一個問題，就是需要等待一段時間再進行調度。因為Yarn是一個非常重吞吐的調度引擎，等待就代表著犧牲了吞吐，這其實對Yarn來說是不可接受的，因此我們在吞吐上也做了一些優化。簡單來說，我們將原生Yarn調度器的單線程改成了多線程版本。

我們把調度器拆分成了讀鎖和寫鎖，對要做的作業包括節點進行了合理分區。然後，我們啟動了多個線程，讓每個線程負責一部分node和作業，這樣就可以達到多線程的效果。在測試環境中，我們的測試結果是在物理資源充裕的情況下，容器的吞吐效率比原來有大概一百倍的提升。

接下來，我將介紹搶佔方面做的優化。Yarn的優勢是更好的利用資源，比如當資源分配完成，A隊列的資源沒有用，而B隊列需要的比較多，B隊列就會把A的資源挪過去用。如果A在此時提交作業，就會發現隊列已經沒有資源了，而B作業沒有運行完，也不會釋放給A，這就需要強調搶佔機制，比如把B作業的部分容器殺死從而釋放資源給A。

Yarn 2.6.0本身自帶搶佔功能，但這基本不可用。舉例說明，如果我要申請一個容器，而該容器需要兩個核才能跑起來，但是Yarn自帶的搶佔機制可能會為我在兩台機器上各搶佔一個核，這基本上是不可用的，這在社區的2.8.0版本中做了重構。我們參考了社區2.8.0版本的代碼，在實現重構的基礎上做了一些優化，並降低了搶佔成本，比如Yarn搶佔要麼全開，要麼不開。 搶佔一定會殺死容器，殺死容器肯定會帶來集群的不穩定，我們定製了一些優化策略， 比如不針對AM，不殺運行時間過長的容器，一個作業只殺死一定比例以下的容器，一台機器只殺死一定比例以下的容器，以此來減少搶佔帶來的影響。

除此之外，社區版本的Yarn在節點達到5000台時，如果發生一次切主，這個集群很可能就會掛掉。對此，我們主要做了兩件事情，一是設置安全模式，保證其在穩定之後再進行調度;二是分析找到所有不必要事件並直接切掉，這讓我們的事件總量下降了800倍左右。 以上措施讓集群可以輕鬆應對萬台機器。

在實際生產環境中，我們很多地方都需要考慮高穩定性。對此，我們做得第一件事是動態 Reservation for Fair Scheduler ，因為一個突然之間需要特別大資源的應用很可能會把整個集群卡住，導致其他應用沒有資源可用。我們的解決方案是能夠Reservation一個上限，並與應用存活時間進行綁定，這可以有效避免整個集群資源被某一個應用全部用掉，而導致其他應用程序沒有資源可用。

此處，我們同樣進行了優化，我們將label信息存儲到了 ZKRMStateStore 中。ZKRMStateStore是原生Yarn用來存放application信息的地方。在Yarn切主時，ZKRMStateStore會把application讀出來，但是原生Yarn會把label放到HDFS之上，這就會讓Yarn對HDFS產生強依賴，當HDFS服務不可用，Yarn一定會掛掉。為了解除強依賴，我們把label信息存放到ZKRMStateStore 中。當切主時，load應用信息的同時會把label信息也load得出來，降低切主時間總長度。並徹底擺脫對HDFS的強依賴，即便HDFS服務掛了，也不會影響Yarn框架的使用。另外，我們修改了容器log的部分策略，比如上傳策略，主要目的是增強功能穩定性，同時降低對HDFS服務的壓力。

當集群規模較大時，用戶上傳的作業、容器log等信息會把HDFS目錄打滿，因為HDFS目錄有默認的存儲上限。針對該情況，我們做了一個哈希希望可以分得更均勻。此外，我們會根據Resource Manager里的壓力負載將NodeManager心跳設置為動態， 如果Resource Manager負載較大，它會自動地把Node Manager的心跳時間設置得更大一些，這樣可以減少一些事件，並減少對Resource Manager的影響。當集群規模較大時，經常會出現Yarn集群與HDFS集群不匹配的情況，比如Yarn集群是ABCD，HDFS集群是BCDE。當按照本地性策略申請資源時，由於沒有版本的計算結點，會等待直到超時退出，我們提前把這些信息保存下來，如果申請的資源在集群內一開始就不存在，我們直接降級成any任何節點都可以。

接下來，我將介紹資源隔離相關操作。更好得資源隔離可以提高在一台機器上同時運行多個作業的效率，我們使用了一種叫NUMA awareness 的技術來做節點內的資源隔離。傳統的資源隔離主要通過Cgroup限制使用率。當多個作業同時運行在一台機器上時，雖然作業跑在不同的CPU和內存上，但要共用系統匯流排帶寬和CPU緩存，作業交替運行刷緩存，導致CPU緩存基本就是不可用的狀態。

NUMA是一種感知架構，對CPU和內存進行綁定，他們之間有獨立的帶寬，如果一個CPU訪問自己的內存，速度會很快，反之訪問其他CPU的內存，速度會很慢。Node Manager起作業時，會直接對作業的CPU和內存進行綁定，這樣會有更好的隔離性。最終，我們的生產環境在實測中實現了部分場景15%的效率提升。

接下來，我將介紹對流式作業的支持。因為Yarn起初設計是為了批處理作業，為了更高的吞吐而設計的，雖然後期也可以支持流式作業，但是Yarn天生並不適合跑流式作業。前文提到，我們公司目前有30%的機器在運行流式作業，因此，我們在這方面做了很多努力，我主要分享兩個事情：一是埠危機，因為流式作業經常與在線服務交互，期間會頻繁建立或斷開連接。受限於TCP連接斷開機制，如果發生意外中斷，系統會等待一段時間才徹底把埠釋放，這樣可以保證它的完備性，比如經常在Node Manager里重啟時會發現埠被佔用，經驗證又發現埠未被佔用。要想解決該問題，一開始就需要考慮特定服務的埠需提前做好預留。因為流式作業需要長時間運行，因此對外部運行時環境依賴是很苛刻的，比如Yarn的Linux container Executor等腳本運行時都需要依賴外部文件，一旦這些文件出問題，就會導致Manager不正常，甚至其上的所有流式容器全部掛掉。

我們還做了容器log的實時查看功能。 在Yarn的設計里，日誌聚合在作業結束後才開始，但流式作業可以認為是不結束的，因此就永遠不會記錄容器log，這也導致用戶查看log不是很方便，我們讓用戶在作業運行過程中通過配置實時把日誌打到Kafka，再到ES進行索引，索引完之後，用戶可以實時在Kibana頁面對作業進行檢索。

接下來，我將介紹我們針對多區域、多數據中心和多集群場景做的事情。首先，我們設計了統一UI，包含用戶所有集群的Job、label和隊列，同時還進行了一些隊列管理，換句話說，所有集群隊列都可以在UI進行管理。其次，我們做了統一hadoop client，這與社區的區別是什麼呢?用戶可以在配置中寫一個簡單的conf，指明集群的名字，這個作業就會被提交到特定集群。

在Yarn的使用過程中，用戶申請資源是一個很頭疼的問題，一般用戶都會提的比較大，這會造成很多資源浪費，這些資源並沒有被真實使用。我們針對這個問題做了兩方面努力。首先，我們使用了Dtop，Dtop會實時收集所有容器的物理資源使用情況。

當然，物理資源不僅局限於CPU和內存，還有磁碟IO、網路IO等。統計完之後，我們會有一個流式作業把這些信息實時聚合。之後，用戶可以在web界面上看到自己應用的資源使用情況，也可以查看單個容器的資源情況，或者整個隊列的資源情況。

在數據的基礎上，我們做了更進一步的實現——動態容器資源調整。首先，我們在Yarn上做了一些開發，因為社區最新版其實已經支持容器動態調整，也就是運行時。我們對這部分功能進行了改進，做了一個比較簡單實用的版本。我們通過Dtop對數據進行實時處理，通過Resource預估器對所有應用信息進行實時聚合。如果用戶提交一個作業，重啟時，Resource預估器可以表明現在需要申請的資源數。在作業運行過程中，它也可以調整流式作業在不同時段，比如高峰期的資源分配。

在易用性層面，我們做了三件事情。首先，我們做了default隊列，用戶可以配置默認隊列，之後提交作業時默認到該隊列。其次，我們做了異常情況通知，當用戶作業或者運行時出現問題，我們都可以通過內部通訊軟體實時把信息反饋給用戶。我們重新設計了Yarn UI，聚合了全球所有集群label、job等信息，用戶可以非常方便的通過一個入口查看所有信息。最後，Troubleshooting支持在易用性里也是一個非常重要的點。由於每天的作業量很大，作業運行失敗的情況時有發生，我們需要快速定位並解決問題。

首先，我介紹一下Btrace工具。該工具主要為了應用源追蹤。比如，我在K8S里起一個在線服務，這在特定場景下會提交一個Hive的SQL語句，該語句又會被翻譯成MapReduce的job提交到Yarn之上，Yarn上的MapReduce的job又會訪問HDFS服務，訪問HDFS服務很可能會把離線帶寬打滿，此時，通過Btrace工具，我們可以在HDFS中快速定位訪問用戶的信息。

接下來是WebShell 工具，該工具主要是為了快速進入容器運行時環境。在原生Yarn容器界面，我們加了一個按鈕叫WebShell，用戶點擊該按鈕就會打開一個瀏覽器頁面，該頁面顯示容器的本地運行環境。這樣，用戶就可以快速在該環境中檢查本地文件信息狀態是否正常。

關於Troubleshooting，還有兩件事情，其一是Reservation可視化，可以讓隊列非常清楚是否因為Reservation而導致資源不足;其二是History Server ，因為Yarn原生History Server 受限於ZK中Znode的大小，一般可存最近的一萬個作業，這對於我們集群而言是遠遠不夠的。為了解決該問題，我們把作業信息實時拉到外部存儲，我們能夠為用戶提供七天的歷史作業查看情況。

除此之外，我們還開發了一些非常有意思的工具，比如Truman。Truman的目標是使用真實的RM，在其它機器上虛構出NM和應用，並且這些NM跟應用表現的與真實環境一樣，只是資源使用非常少，甚至基本不佔用資源。這樣，我們就可以較低成本測試RM的性能。另一個工具是ClusterManager，可中心化服務管理節點上聲明的計算資源和label信息，該功能在社區有實現，但會把權力下放給Node Manager。

另一個是LogIndexService，可以實時抓取用戶作業的容器log，並傳遞到Kafka和ES，最後通過Kibana頁面展示。我們的運維平台工具叫YAOP，這是一個全球多機房統一運維管理平台，負責管理所有用戶的job、隊列、label和節點信息。

4、Future Works

未來，我們需要做兩件事情。 一是Federation機制，我們需要通過該機制對多個同一IDC 集群進行聯合，為用戶提供統一視圖，並提高跨集群資源利用率;二是Docker on yarn ，為了更好地實現一台物理機之內的資源隔離。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！