TalkingData的Spark On Kubernetes實(shí)踐是怎樣的

這篇文章給大家介紹TalkingData的Spark On Kubernetes實(shí)踐是怎樣的，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

眾所周知，Spark是一個快速、通用的大規(guī)模數(shù)據(jù)處理平臺，和Hadoop的MapReduce計算框架類似。但是相對于MapReduce，Spark憑借其可伸縮、基于內(nèi)存計算等特點(diǎn)，以及可以直接讀寫Hadoop上任何格式數(shù)據(jù)的優(yōu)勢，使批處理更加高效，并有更低的延遲。實(shí)際上，Spark已經(jīng)成為輕量級大數(shù)據(jù)快速處理的統(tǒng)一平臺。
Spark作為一個數(shù)據(jù)計算平臺和框架，更多的是關(guān)注Spark Application的管理，而底層實(shí)際的資源調(diào)度和管理更多的是依靠外部平臺的支持：

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

如何搭建生產(chǎn)可用的Kubernetes集群

部署

目前市面上有很多搭建Kubernetes的方法，比如Scratch、Kubeadm、Minikube或者各種托管方案。因?yàn)槲覀冃枰唵慰焖俚卮罱üδ茯?yàn)證集群，所以選擇了Kubeadm作為集群的部署工具。部署步驟很簡單，在master上執(zhí)行：

kubeadm init

在node上執(zhí)行：

kubeadm join --token  : --discovery-token-ca-cert-hash sha256:

具體配置可見官方文檔：https://kubernetes.io/docs/setup/independent/create-cluster-kubeadm/。
需要注意的是由于國內(nèi)網(wǎng)絡(luò)限制，很多鏡像無法從k8s.gcr.io獲取，我們需要將之替換為第三方提供的鏡像，比如：https://hub.docker.com/u/mirrorgooglecontainers/。

網(wǎng)絡(luò)

Kubernetes網(wǎng)絡(luò)默認(rèn)是通過CNI實(shí)現(xiàn)，主流的CNI plugin有：Linux Bridge、MACVLAN、Flannel、Calico、Kube-router、Weave Net等。Flannel主要是使用VXLAN tunnel來解決pod間的網(wǎng)絡(luò)通信，Calico和Kube-router則是使用BGP。由于軟VXLAN對宿主機(jī)的性能和網(wǎng)絡(luò)有不小的損耗，BGP則對硬件交換機(jī)有一定的要求，且我們的基礎(chǔ)網(wǎng)絡(luò)是VXLAN實(shí)現(xiàn)的大二層，所以我們最終選擇了MACVLAN。
CNI MACVLAN的配置示例如下：

{
  "name": "mynet",
  "type": "macvlan",
  "master": "eth0",
  "ipam": {
    "type": "host-local",
    "subnet": "10.0.0.0/17",
    "rangeStart": "10.0.64.1",
    "rangeEnd": "10.0.64.126",
    "gateway": "10.0.127.254",
    "routes": [
      {
        "dst": "0.0.0.0/0"
      },
      {
        "dst": "10.0.80.0/24",
        "gw": "10.0.0.61"
      }
    ]
  }
}

Pod subnet是10.0.0.0/17，實(shí)際pod ip pool是10.0.64.0/20。cluster cidr是10.0.80.0/24。我們使用的IPAM是host-local，規(guī)則是在每個Kubernetes node上建立/25的子網(wǎng)，可以提供126個IP。我們還配置了一條到cluster cidr的靜態(tài)路由10.0.80.0/24，網(wǎng)關(guān)是宿主機(jī)。這是因?yàn)槿萜髟趍acvlan配置下egress并不會通過宿主機(jī)的iptables，這點(diǎn)和Linux Bridge有較大區(qū)別。在Linux Bridge模式下，只要指定內(nèi)核參數(shù)net.bridge.bridge-nf-call-iptables = 1，所有進(jìn)入bridge的流量都會通過宿主機(jī)的iptables。經(jīng)過分析kube-proxy，我們發(fā)現(xiàn)可以使用KUBE-FORWARD這個chain來進(jìn)行pod到service的網(wǎng)絡(luò)轉(zhuǎn)發(fā)：

-A FORWARD -m comment --comment "kubernetes forward rules" -j KUBE-FORWARD
-A KUBE-FORWARD -m comment --comment "kubernetes forwarding rules" -m mark --mark 0x4000/0x4000 -j ACCEPT
-A KUBE-FORWARD -s 10.0.0.0/17 -m comment --comment "kubernetes forwarding conntrack pod source rule" -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT
-A KUBE-FORWARD -d 10.0.0.0/17 -m comment --comment "kubernetes forwarding conntrack pod destination rule" -m conntrack --ctstate RELATED,ESTABLISHED -j ACCEPT

最后通過KUBE-SERVICES使用DNAT到后端的pod。pod訪問其他網(wǎng)段的話，就通過物理網(wǎng)關(guān)10.0.127.254。
還有一個需要注意的地方是出于kernel security的考慮，link物理接口的macvlan是無法直接和物理接口通信的，這就導(dǎo)致容器并不能將宿主機(jī)作為網(wǎng)關(guān)。我們采用了一個小技巧，避開了這個限制。我們從物理接口又創(chuàng)建了一個macvlan，將物理IP移到了這個接口上，物理接口只作為網(wǎng)絡(luò)入口：

$ cat /etc/sysconfig/network-scripts/ifcfg-eth0
DEVICE=eth0
IPV6INIT=no
BOOTPROTO=none
$ cat /etc/sysconfig/network-scripts/ifcfg-macvlan
DEVICE=macvlan
NAME=macvlan
BOOTPROTO=none
ONBOOT=yes
TYPE=macvlan
DEVICETYPE=macvlan
DEFROUTE=yes
PEERDNS=yes
PEERROUTES=yes
IPV4_FAILURE_FATAL=no
IPADDR=10.0.0.61
PREFIX=17
GATEWAY=10.0.127.254
MACVLAN_PARENT=eth0
MACVLAN_MODE=bridge

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

Kube-dns

默認(rèn)配置下，Kubernetes使用kube-dns進(jìn)行DNS解析和服務(wù)發(fā)現(xiàn)。但在實(shí)際使用時，我們發(fā)現(xiàn)在pod上通過service domain訪問service總是有5秒的延遲。使用tcpdump抓包，發(fā)現(xiàn)延遲出現(xiàn)在DNS AAAA。進(jìn)一步排查，發(fā)現(xiàn)問題是由于netfilter在conntrack和SNAT時的Race Condition導(dǎo)致。簡言之，DNS A和AAAA記錄請求報文是并行發(fā)出的，這會導(dǎo)致netfilter在_nf_conntrack_confirm時認(rèn)為第二個包是重復(fù)的（因?yàn)橛邢嗤奈逶M），從而丟包。具體可看我提的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，設(shè)置hostNetwork=true使用宿主機(jī)網(wǎng)絡(luò)提供DNS服務(wù)。因?yàn)槲覀兊幕A(chǔ)網(wǎng)絡(luò)是大二層，所以pod和node可以直接通信，這就避免了conntrack和SNAT。

Spark與Kubernetes集成

由于Spark的抽象設(shè)計，我們可以使用第三方資源管理平臺調(diào)度和管理Spark作業(yè)，比如Yarn、Mesos和Kubernetes。目前官方有一個experimental項(xiàng)目，可以將Spark運(yùn)行在Kubernetes之上：https://spark.apache.org/docs/latest/running-on-kubernetes.html。

基本原理

當(dāng)我們通過spark-submit將Spark作業(yè)提交到Kubernetes集群時，會執(zhí)行以下流程：

• Spark在Kubernetes pod中創(chuàng)建Spark driver

• Driver調(diào)用Kubernetes API創(chuàng)建executor pods，executor pods執(zhí)行作業(yè)代碼

• 計算作業(yè)結(jié)束，executor pods回收并清理

• driver pod處于completed狀態(tài)，保留日志，直到Kubernetes GC或者手動清理

先決條件

• Spark 2.3+

• Kubernetes 1.6+

• 具有Kubernetes pods的list, create, edit和delete權(quán)限

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

如何集成

Docker鏡像

由于Spark driver和executor都運(yùn)行在Kubernetes pod中，并且我們使用Docker作為container runtime enviroment，所以首先我們需要建立Spark的Docker鏡像。
在Spark distribution中已包含相應(yīng)腳本和Dockerfile，可以通過以下命令構(gòu)建鏡像：

$ ./bin/docker-image-tool.sh -r <repo> -t my-tag build
$ ./bin/docker-image-tool.sh -r <repo> -t my-tag push

提交作業(yè)

在構(gòu)建Spark鏡像后，我們可以通過以下命令提交作業(yè)：

$ bin/spark-submit \
    --master k8s://https://: \
    --deploy-mode cluster \
    --name spark-pi \
    --class org.apache.spark.examples.SparkPi \
    --jars https://path/to/dependency1.jar,https://path/to/dependency2.jar
    --files hdfs://host:port/path/to/file1,hdfs://host:port/path/to/file2
    --conf spark.executor.instances=5 \
    --conf spark.kubernetes.container.image= \
    https://path/to/examples.jar

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

$ kubectl cluster-info
Kubernetes master is running at http://127.0.0.1:6443

Spark的作業(yè)代碼和依賴，我們可以在--jars、--files和最后位置指定，協(xié)議支持http、https和HDFS。
執(zhí)行提交命令后，會有以下輸出：

任務(wù)結(jié)束，會輸出：

訪問Spark Driver UI

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

$ kubectl port-forward <driver-pod-name> 4040:4040

執(zhí)行完后通過http://localhost:4040訪問。

訪問日志

Spark的所有日志都可以通過Kubernetes API和kubectl CLI進(jìn)行訪問：

$ kubectl -n=<namespace> logs -f <driver-pod-name>

如何實(shí)現(xiàn)租戶和資源隔離

Kubernetes Namespace

在Kubernetes中，我們可以使用namespace在多用戶間實(shí)現(xiàn)資源分配、隔離和配額。Spark On Kubernetes同樣支持配置namespace創(chuàng)建Spark作業(yè)。
首先，創(chuàng)建一個Kubernetes namespace：

$ kubectl create namespace spark

由于我們的Kubernetes集群使用了RBAC，所以還需創(chuàng)建serviceaccount和綁定role：

$ kubectl create serviceaccount spark -n spark
$ kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=spark:spark --namespace=spark

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

$ bin/spark-submit \
    --conf spark.kubernetes.authenticate.driver.serviceAccountName=spark \
    --conf spark.kubernetes.namespace=spark \
    ...

資源隔離

考慮到我們Spark作業(yè)的一些特點(diǎn)和計算資源隔離，前期我們還是選擇了較穩(wěn)妥的物理隔離方案。具體做法是為每個組提供單獨(dú)的Kubernetes namespace，計算任務(wù)都在各自namespace里提交。計算資源以物理機(jī)為單位，折算成cpu和內(nèi)存，納入Kubernetes統(tǒng)一管理。在Kubernetes集群里，通過node label和PodNodeSelector將計算資源和namespace關(guān)聯(lián)。從而實(shí)現(xiàn)在提交Spark作業(yè)時，計算資源總是選擇namespace關(guān)聯(lián)的node。
具體做法如下：
1、創(chuàng)建node label

$ kubectl label nodes <node_name> spark:spark

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

$ cat /etc/kubernetes/manifests/kube-apiserver.yaml
apiVersion: v1
kind: Pod
metadata:
  annotations:
    scheduler.alpha.kubernetes.io/critical-pod: ""
  creationTimestamp: null
  labels:
    component: kube-apiserver
    tier: control-plane
  name: kube-apiserver
  namespace: kube-system
spec:
  containers:
  - command:
    - kube-apiserver
    - --secure-port=6443
    - --proxy-client-cert-file=/etc/kubernetes/pki/front-proxy-client.crt
    - --admission-control=Initializers,NamespaceLifecycle,LimitRanger,ServiceAccount,DefaultStorageClass,DefaultTolerationSeconds,NodeRestriction,ResourceQuota,MutatingAdmissionWebhook,ValidatingAdmissionWebhook,PodNodeSelector
...

3、配置PodNodeSelector
在namespace的annotations中添加scheduler.alpha.kubernetes.io/node-selector: spark=spark。

apiVersion: v1
kind: Namespace
metadata:
 annotations:
   scheduler.alpha.kubernetes.io/node-selector: spark=spark
 name: spark

完成以上配置后，可以通過spark-submit測試結(jié)果：

$ spark-submit
--conf spark.kubernetes.authenticate.driver.serviceAccountName=spark 
--conf spark.kubernetes.namespace=spark 
--master k8s://https://xxxx:6443     
--deploy-mode cluster     
--name spark-pi     
--class org.apache.spark.examples.SparkPi     
--conf spark.executor.instances=5     
--conf spark.kubernetes.container.image=xxxx/library/spark:v2.3     
http://xxxx:81/spark-2.3.0-bin-hadoop2.7/examples/jars/spark-examples_2.11-2.3.0.jar

我們可以看到，Spark作業(yè)全分配到了關(guān)聯(lián)的hadooptest-001到003三個node上。

待解決問題

Kubernetes HA

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

日志

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

$ kubectl -n=<namespace> logs -f <driver-pod-name>

收集和聚合日志，我們后面會和ES結(jié)合。
監(jiān)控
我們TalkingData內(nèi)部有自己的監(jiān)控平臺OWL[2]（已開源），未來我們計劃編寫metric plugin，將Kubernetes接入OWL中。
混合部署
為了保證Spark作業(yè)時刻有可用的計算資源，我們前期采用了物理隔離的方案。顯而易見，這種方式大幅降低了物理資源的使用率。下一步我們計劃采用混部方案，通過以下三種方式實(shí)現(xiàn)：

• 將HDFS和Kubernetes混合部署

• 為Spark作業(yè)和Kubernetes node劃分優(yōu)先級，在低優(yōu)先級的node上同時運(yùn)行一些無狀態(tài)的其他生產(chǎn)服務(wù)

• 利用云實(shí)現(xiàn)資源水平擴(kuò)展，以防止資源突增

資源擴(kuò)展

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

• 混合部署

• 資源超賣

這會導(dǎo)致運(yùn)行資源大于實(shí)際物理資源的情況（我稱之為資源擠兌）。一種做法是給資源劃分等級，優(yōu)先保證部分等級的資源供給。另一種做法是實(shí)現(xiàn)資源的水平擴(kuò)展，動態(tài)補(bǔ)充可用資源，并在峰值過后自動釋放。我在另一篇文章中闡述了這種設(shè)計理念：https://xiaoxubeii.github.io/articles/k8s-on-cloud/。
TalkingData有自研的多云管理平臺，我們的解決方法是實(shí)現(xiàn)單獨(dú)的Kubernetes tdcloud-controller-manager作為資源的provider和manager，通過TalkingData OWL監(jiān)控告警，實(shí)現(xiàn)資源的水平擴(kuò)展。

關(guān)于TalkingData的Spark On Kubernetes實(shí)踐是怎樣的就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。