TalkingData的Spark On Kubernetes實踐

眾所周知，Spark是一個快速、通用的大規模數據處理平台，和Hadoop的MapReduce計算框架類似。但是相對於MapReduce，Spark憑藉其可伸縮、基於內存計算等特點，以及可以直接讀寫Hadoop上任何格式數據的優勢，使批處理更加高效，並有更低的延遲。實際上，Spark已經成為輕量級大數據快速處理的統一平台。

Spark作為一個數據計算平台和框架，更多的是關注Spark Application的管理，而底層實際的資源調度和管理更多的是依靠外部平台的支持：

Spark官方支持四種Cluster Manager：Spark standalone cluster manager、Mesos、YARN和Kubernetes。由於我們TalkingData是使用Kubernetes作為資源的調度和管理平台，所以Spark On Kubernetes對於我們是最好的解決方案。

如何搭建生產可用的Kubernetes集群

部署

目前市面上有很多搭建Kubernetes的方法，比如Scratch、Kubeadm、Minikube或者各種託管方案。因為我們需要簡單快速地搭建功能驗證集群，所以選擇了Kubeadm作為集群的部署工具。部署步驟很簡單，在master上執行：

在node上執行：

具體配置可見官方文檔：https://kubernetes.io/docs/setup/independent/create-cluster-kubeadm/。

需要注意的是由於國內網路限制，很多鏡像無法從k8s.gcr.io獲取，我們需要將之替換為第三方提供的鏡像，比如：https://hub.docker.com/u/mirrorgooglecontainers/。

網路

Kubernetes網路默認是通過CNI實現，主流的CNI plugin有：Linux Bridge、MACVLAN、Flannel、Calico、Kube-router、Weave Net等。Flannel主要是使用VXLAN tunnel來解決pod間的網路通信，Calico和Kube-router則是使用BGP。由於軟VXLAN對宿主機的性能和網路有不小的損耗，BGP則對硬體交換機有一定的要求，且我們的基礎網路是VXLAN實現的大二層，所以我們最終選擇了MACVLAN。

CNI MACVLAN的配置示例如下：

Pod subnet是10.0.0.0/17，實際pod ip pool是10.0.64.0/20。cluster cidr是10.0.80.0/24。我們使用的IPAM是host-local，規則是在每個Kubernetes node上建立/25的子網，可以提供126個IP。我們還配置了一條到cluster cidr的靜態路由10.0.80.0/24，網關是宿主機。這是因為容器在macvlan配置下egress並不會通過宿主機的iptables，這點和Linux Bridge有較大區別。在Linux Bridge模式下，只要指定內核參數net.bridge.bridge-nf-call-iptables = 1，所有進入bridge的流量都會通過宿主機的iptables。經過分析kube-proxy，我們發現可以使用KUBE-FORWARD這個chain來進行pod到service的網路轉發：

最後通過KUBE-SERVICES使用DNAT到後端的pod。pod訪問其他網段的話，就通過物理網關10.0.127.254。

還有一個需要注意的地方是出於kernel security的考慮，link物理介面的macvlan是無法直接和物理介面通信的，這就導致容器並不能將宿主機作為網關。我們採用了一個小技巧，避開了這個限制。我們從物理介面又創建了一個macvlan，將物理IP移到了這個介面上，物理介面只作為網路入口：

這樣兩個macvlan是可以互相通信的。

Kube-dns

默認配置下，Kubernetes使用kube-dns進行DNS解析和服務發現。但在實際使用時，我們發現在pod上通過service domain訪問service總是有5秒的延遲。使用tcpdump抓包，發現延遲出現在DNS AAAA。進一步排查，發現問題是由於netfilter在conntrack和SNAT時的Race Condition導致。簡言之，DNS A和AAAA記錄請求報文是並行發出的，這會導致netfilter在_nf_conntrack_confirm時認為第二個包是重複的（因為有相同的五元組），從而丟包。具體可看我提的issue：https://github.com/kubernetes/kubernetes/issues/62628。一個簡單的解決方案是在/etc/resolv.conf中增加options single-request-reopen，使DNS A和AAAA記錄請求報文使用不同的源埠。我提的PR在：https://github.com/kubernetes/kubernetes/issues/62628，大家可以參考。我們的解決方法是不使用Kubernetes service，設置hostNetwork=true使用宿主機網路提供DNS服務。因為我們的基礎網路是大二層，所以pod和node可以直接通信，這就避免了conntrack和SNAT。

Spark與Kubernetes集成

由於Spark的抽象設計，我們可以使用第三方資源管理平台調度和管理Spark作業，比如Yarn、Mesos和Kubernetes。目前官方有一個experimental項目，可以將Spark運行在Kubernetes之上：https://spark.apache.org/docs/latest/running-on-kubernetes.html。

基本原理

當我們通過spark-submit將Spark作業提交到Kubernetes集群時，會執行以下流程：

Spark在Kubernetes pod中創建Spark driver

Driver調用Kubernetes API創建executor pods，executor pods執行作業代碼

計算作業結束，executor pods回收並清理

driver pod處於completed狀態，保留日誌，直到Kubernetes GC或者手動清理

先決條件

Spark 2.3+

Kubernetes 1.6+

具有Kubernetes pods的list, create, edit和delete許可權

Kubernetes集群必須正確配置Kubernetes DNS[1]

如何集成

Docker鏡像

由於Spark driver和executor都運行在Kubernetes pod中，並且我們使用Docker作為container runtime enviroment，所以首先我們需要建立Spark的Docker鏡像。

在Spark distribution中已包含相應腳本和Dockerfile，可以通過以下命令構建鏡像：

提交作業

在構建Spark鏡像後，我們可以通過以下命令提交作業：

其中，Spark master是Kubernetes api server的地址，可以通過以下命令獲取：

Spark的作業代碼和依賴，我們可以在--jars、--files和最後位置指定，協議支持http、https和HDFS。

執行提交命令後，會有以下輸出：

任務結束，會輸出：

訪問Spark Driver UI

我們可以在本地使用kubectl port-forward訪問Driver UI：

執行完後通過http://localhost:4040訪問。

訪問日誌

Spark的所有日誌都可以通過Kubernetes API和kubectl CLI進行訪問：

如何實現租戶和資源隔離

Kubernetes Namespace

在Kubernetes中，我們可以使用namespace在多用戶間實現資源分配、隔離和配額。Spark On Kubernetes同樣支持配置namespace創建Spark作業。

首先，創建一個Kubernetes namespace：

由於我們的Kubernetes集群使用了RBAC，所以還需創建serviceaccount和綁定role：

並在spark-submit中新增以下配置：

資源隔離

考慮到我們Spark作業的一些特點和計算資源隔離，前期我們還是選擇了較穩妥的物理隔離方案。具體做法是為每個組提供單獨的Kubernetes namespace，計算任務都在各自namespace里提交。計算資源以物理機為單位，折算成cpu和內存，納入Kubernetes統一管理。在Kubernetes集群里，通過node label和PodNodeSelector將計算資源和namespace關聯。從而實現在提交Spark作業時，計算資源總是選擇namespace關聯的node。

具體做法如下：

1、創建node label

2、開啟Kubernetes admission controller

我們是使用kubeadm安裝Kubernetes集群，所以修改/etc/kubernetes/manifests/kube-apiserver.yaml，在--admission-control後添加PodNodeSelector。

3、配置PodNodeSelector

完成以上配置後，可以通過spark-submit測試結果：

我們可以看到，Spark作業全分配到了關聯的hadooptest-001到003三個node上。

待解決問題

Kubernetes HA

Kubernetes的集群狀態基本都保存在etcd中，所以etcd是HA的關鍵所在。由於我們目前還處在半生產狀態，HA這方面未過多考慮。有興趣的同學可以查看：https://kubernetes.io/docs/setup/independent/high-availability/。

日誌

在Spark On Yarn下，可以開啟yarn.log-aggregation-enable將日誌收集聚合到HDFS中，以供查看。但是在Spark On Kubernetes中，則缺少這種日誌收集機制，我們只能通過Kubernetes pod的日誌輸出，來查看Spark的日誌：

收集和聚合日誌，我們後面會和ES結合。

監控

我們TalkingData內部有自己的監控平台OWL[2]（已開源），未來我們計劃編寫metric plugin，將Kubernetes接入OWL中。

混合部署

為了保證Spark作業時刻有可用的計算資源，我們前期採用了物理隔離的方案。顯而易見，這種方式大幅降低了物理資源的使用率。下一步我們計劃採用混部方案，通過以下三種方式實現：

將HDFS和Kubernetes混合部署

為Spark作業和Kubernetes node劃分優先順序，在低優先順序的node上同時運行一些無狀態的其他生產服務

利用雲實現資源水平擴展，以防止資源突增

資源擴展

在採用以下兩種方法增加資源使用率時，集群可能會面臨資源短缺和可用性的問題：

混合部署

資源超賣

這會導致運行資源大於實際物理資源的情況（我稱之為資源擠兌）。一種做法是給資源劃分等級，優先保證部分等級的資源供給。另一種做法是實現資源的水平擴展，動態補充可用資源，並在峰值過後自動釋放。我在另一篇文章中闡述了這種設計理念：https://xiaoxubeii.github.io/articles/k8s-on-cloud/。

TalkingData有自研的多雲管理平台，我們的解決方法是實現單獨的Kubernetes tdcloud-controller-manager作為資源的provider和manager，通過TalkingData OWL監控告警，實現資源的水平擴展。

Q&A

Q：為什麼使用node label進行資源隔離，而不使用ResourceQuota對多租戶進行資源隔離？

A：由於我們很多大數據計算作業對SLA有很高的要求，並且Docker實際上對很多應用的資源限制都支持的不好。所以我們前期為了穩妥，還是對計算資源進行了物理隔離。

Q：比如flannel是把flannel參數傳給Docker，一種用CNI插件，他們有何差別？

A：實際上CNI是Kubernetes的標準網路介面，而flannel是實現Pod間通信的網路插件。CNI中有兩類插件：一個是ipam，一個是network plugins。flannel屬於後者，是可以納入CNI管理的。

Q：這裡的多租戶隔離，只提到任務執行過程的調度，那對於不同租戶的任務提交，狀態監控，結果呈現如何實現隔離的？

A：不同的租戶對應不同的Kubernetes namespace，所以自然實現了任務提交和狀態監控的隔離。至於計算結果，我們以往是單純用hdfs path做隔離。我們目前內部有大數據平台，那裡真正實現了多租戶。

Q：Spark On Kubernetes這種方式為開發人員增加了難度，不像其他的集群方案，開發人員除了要會 Spark還要會Kubernetes，請問怎麼推？

A：實際上Spark On Kubernetes對大數據開發人員是透明的，任務的提交方式並沒有改變，只是加了一些額外的option。並且我們上層是有統一的大數據平台，進行作業提交。

Q：在使用HDFS存儲資源下，如果不使用Spark的數據本地性，大量數據之間的傳輸，map和reduce操作是否會影響Spark的計算性能呢？

A：個人認為肯定會有影響，因為每次從HDFS讀取，會帶來巨大的網路流量。但是我本身對Spark的數據本地性沒有什麼研究。後期我們計劃將HDFS和Kubernetes混部，將數據盡量靠近計算節點，不知道這種方式能否緩解這個問題。同時，我們還可以使用Spark on Kubernetes的external-shuffle-service，從而使用hostpath volume將shuffle和executor pods打通。

Q：Spark會作為哪種資源部署方式部署？Deployment還是StatefulSet？或者其他？Spark在生產環境上部署有需要什麼資源配置？能否分享下TalkingData的生產環境是如何分配Spark資源的？

A：Spark On Kubernetes實際就是創建了Spark driver headless service，然後創建Spark driver pod，最後由driver創建executors pods。上述分享中我也提到了，目前我們還是以物理機作為spark資源分配的單位。

Q：Yarn vs Kubernetes優缺點?

A：我們以前的Spark就是採用Spark On Yarn的方式，不過我對Yarn不是非常了解。之所以採用Kubernetes是因為，我們想統一底層的資源調度平台。但是Yarn目前還是和Hadoop生態強耦合的。

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