在Kubernetes中运行Spark集群

在Kubernetes中运行Spark集群

• 注意，Spark 2.3已经正式发布，原生支持Kubernetes，这里的内容已通过时。
  • 参考 http://www.javashuo.com/article/p-kxmzrnjw-e.html

概述

Spark是新一代分布式内存计算框架，Apache开源的顶级项目。相比于Hadoop Map-Reduce计算框架，Spark将中间计算结果保留在内存中，速度提高10~100倍；同时它还提供更丰富的算子，采用弹性分布式数据集(RDD)实现迭代计算，更好地适用于数据挖掘、机器学习算法，极大提高开发效率。

Docker是轻量级虚拟化容器技术，具备轻便性、隔离性、一致性等特色，能够极大简化开发者的部署运维流程，下降服务器成本。

Kubernetes是Google开源的容器集群管理系统，提供应用部署、维护、 扩展等功能，可以方便地管理大规模跨主机的容器应用。

    相比于在物理机上部署，在Kubernetes集群上部署Spark集群，具备如下优点：

• 快速部署：安装1000台级别的Spark集群，在Kubernetes集群上只需设定worker副本数目replicas=1000，便可一键部署。
• 快速升级：升级Spark版本，只需替换Spark镜像，一键升级。
• 弹性伸缩：须要扩容、缩容时，自动修改worker副本数目replicas便可。
• 高一致性：各个Kubernetes节点上运行的Spark环境一致、版本一致
• 高可用性：若是Spark所在的某些node或pod死掉，Kubernetes会自动将计算任务，转移到其余node或建立新pod。
• 强隔离性：经过设定资源配额等方式，可与WebService应用部署在同一集群，提高机器资源使用效率，从而下降服务器成本。

参考文档：https://github.com/kubernetes/kubernetes/tree/master/examples/spark

接下来咱们将介绍，如何使用 Docker和Kubernetes建立Spark集群（其中包含1个Spark-master和N个Spark-worker）。

首先须要准备如下工具：

• 安装并运行 kubernetes 集群
• 安装 kubectl 命令行工具 

1、构建Docker镜像

1. 从源码build：

下载：https://github.com/kubernetes/application-images/blob/master/spark

docker build -t index.caicloud.io/spark:1.5.2 .
docker build -t index.caicloud.io/zeppelin:0.5.6 zeppelin/

链接国外网络较慢，构建镜像的时间有些长，请耐心等待……

2. 从镜像仓库pull：

a)  才云科技的镜像仓库（index.caicloud.io）

docker login index.caicloud.io 
docker pull index.caicloud.io/spark:1.5.2 
docker pull index.caicloud.io/zeppelin:0.5.6

b) DockerHub的镜像仓库（index.docker.io）

docker login index.docker.io
docker pull index.docker.io/caicloud/spark:1.5.2
docker pull index.docker.io/caicloud/zeppelin:0.5.6

 

2、在Kubernetes上建立Spark集群

首先下载建立Kubernetes应用所需的Yaml文件: https://github.com/caicloud/public/tree/master/spark

第一步：建立命名空间namespace

    Kubernetes经过命名空间，将底层的物理资源划分红若干个逻辑的“分区”，然后续全部的应用、容器都是被部署在一个具体的命名空间里。每一个命名空间能够设置独立的资源配额，保证不一样命名空间中的应用不会相互抢占资源。此外，命名空间对命名域实现了隔离，所以两个不一样命名空间里的应用能够起一样的名字。建立命名空间须要编写一个yaml文件：

# namespace/namespace-spark-cluster.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: "spark-cluster"
  labels:
    name: "spark-cluster"

 

建立Namespace:
$ kubectl create -f namespace/namespace-spark-cluster.yaml

查看Namespace:
$ kubectl get ns
NAME          LABELS             STATUS
default       <none>             Active
spark-cluster name=spark-cluster Active

使用Namespace: (${CLUSTER_NAME}和${USER_NAME}可在kubeconfig文件中查看)
$ kubectl config set-context spark --namespace=spark-cluster --cluster=${CLUSTER_NAME} --user=${USER_NAME}
$ kubectl config use-context spark

 

第二步：启动master服务

    先建立Spark-master的ReplicationController，而后建立spark所提供的两个Service(spark-master-service，spark-webui)。让Spark-workers使用spark-master-service来链接Spark-master，而且经过spark-webui来查看集群和任务运行状态。

1) 建立ReplicationController：

    Kubernetes追求高可用设计，经过Replication Controller来保证每一个应用时时刻刻会有指定数量的副本在运行。例如咱们经过编写一个Replication Controller来运行一个nginx应用，就能够在yaml中指定5个默认副本。Kubernetes会自动运行5个nginx副本，并在后期时时对每个副本进行健康检查（能够支持自定义的检查策略）。当发现有副本不健康时，Kubernetes会经过自动重启、迁移等方法，保证nginx会时刻有5个健康的副本在运行。对于spark-master，目前咱们指定其副本数为1 (replicas: 1)；对于spark-worker，咱们指定其副本数为N (replicas: N，N >= 1)。

    spark-master-controller.yaml可参考以下：

# replication-controller/spark-master-controller.yaml
kind: ReplicationController
apiVersion: v1
metadata:
  name: spark-master-controller
spec:
  replicas: 1
  selector:
    component: spark-master
  template:
    metadata:
      labels:
        component: spark-master
    spec:
      containers:
        - name: spark-master
          image: index.caicloud.io/spark:1.5.2
          command: ["/start-master"]
          ports:
            - containerPort: 7077
            - containerPort: 8080
          resources:
            requests:
              cpu: 100m

 

建立Master-ReplicationController：
$ kubectl create -f replication-controller/spark-master-controller.yaml
replicationcontroller "spark-master-controller" created

 

2) 建立Master-Service：

    Kubernetes追求以服务为中心，并推荐为系统中的应用建立对应的Service。以nginx应用为例，当经过Replication Controller建立了多个nginx的实例（容器）后，这些不一样的实例可能运行在不一样的节点上，而且随着故障和自动修复，其IP可能会动态变化。为了保证其余应用能够稳定地访问到nginx服务，咱们能够经过编写yaml文件为nginx建立一个Service，并指定该Service的名称（如nginx-service）；此时，Kubernetes会自动在其内部一个DNS系统中（基于SkyDNS 和etcd实现）为其添加一个A Record, 名字就是 “nginx-service”。随后，其余的应用能够经过 nginx-service来自动寻址到nginx的一个实例（用户能够配置负载均衡策略）。
    spark-master-service.yaml可参考以下：

# service/spark-master-service.yaml
kind: Service
apiVersion: v1
metadata:
  name: spark-master
spec:
  ports:
    - port: 7077
      targetPort: 7077
  selector:
    component: spark-master

 

建立Master-Service:
$ kubectl create -f service/spark-master-service.yaml
service "spark-master"created

 

3) 建立WebUI-Service：

    如上所述，Service会被映射到后端的实际容器应用上，而这个映射是经过Kubernetes的标签以及Service的标签选择器实现的。例如咱们能够经过以下的spark-web-ui.yaml来建立一个WebUI的Service, 而这个Service会经过 “selector: component: spark-master”来把WebUI的实际业务映射到master节点上:

# service/spark-webui.yaml
kind: Service
apiVersion: v1
metadata:
  name: spark-webui
  namespace: spark-cluster
spec:
  ports:
    - port: 8080
      targetPort: 8080
  selector:
    component: spark-master

 

建立spark-webui-service：
$ kubectl create -f service/spark-webui.yaml
service "spark-webui" created

完成建立ReplicationController(rc)、Service(svc)后，检查 Master 是否能运行和访问：

$ kubectl get rc
NAME                      DESIRED   CURRENT   AGE
spark-master-controller   1         1         23h

$ kubectl get svc
NAME           CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE
spark-master   10.254.106.29   <none>        7077/TCP   1d
spark-webui    10.254.66.138   <none>        8080/TCP   18h

$ kubectl get pods
NAME                            READY     STATUS    RESTARTS   AGE
spark-master-controller-b3gbf   1/1       Running   0          23h

确认master正常运行后，再使用Kubernetes proxy链接Spark WebUI：

kubectl proxy --port=8001

而后经过http://localhost:8001/api/v1/proxy/namespaces/spark-cluster/services/spark-webui/查看spark的任务运行状态。

第三步：启动 Spark workers

    Spark workers 启动时须要 Master service处于运行状态，咱们能够经过修改replicas来设定worker数目（好比设定 replicas: 4，便可创建4个Spark Worker）。注意咱们能够为每个worker节点设置了CPU和内存的配额，保证Spark的worker应用不会过分抢占集群中其余应用的资源。

    spark-worker-controller.yaml可参考以下：

# replication-controller/spark-worker-controller.yaml
kind: ReplicationController
apiVersion: v1
metadata:
  name: spark-worker-controller
spec:
  replicas: 4
  selector:
    component: spark-worker
  template:
    metadata:
      labels:
        component: spark-worker
    spec:
      containers:
        - name: spark-worker
          image: index.caicloud.io/spark:1.5.2
          command: ["/start-worker"]
          ports:
            - containerPort: 8081
          resources:
            requests:
              cpu: 100m

建立spark-worker的ReplicationController：

$ kubectl create -f replication-controller/spark-worker-controller.yaml
replicationcontroller"spark-worker-controller" created

 

查看 workers是否正常运行

1.     经过WebUI查看： worker就绪后应该出如今UI中。（这可能须要一些时间来拉取镜像并启动pods）

2.   经过kubectl查询状态（可看到spark-worker都已经正常运行）：

</pre><p><pre name="code" class="plain">$ kubectl get pods
NAME                            READY     STATUS              RESTARTS   AGE
spark-master-controller-b3gbf   1/1       Running             0          1d
spark-worker-controller-ill4z   1/1       Running             1          2h
spark-worker-controller-j29sc   1/1       Running             0          2h
spark-worker-controller-siue2   1/1       Running             0          2h
spark-worker-controller-zd5kb   1/1       Running             0          2h

 

第四步：提交Spark任务

两种方式：Spark-client或Zeppelin。

经过Spark-client，能够利用spark-submit来提交复杂的Python脚本、Java/Scala的jar包代码；

经过Zeppelin，能够直接在命令行或UI编写简单的spark代码。

1.     经过Spark-client提交任务

$ kubectl get pods | grep worker

NAME                            READY     STATUS    RESTARTS   AGE
spark-worker-controller-1h0l7   1/1       Running   0          4h
spark-worker-controller-d43wa   1/1       Running   0          4h
spark-worker-controller-ka78h   1/1       Running   0          4h
spark-worker-controller-sucl7   1/1       Running   0          4h

$ kubectl exec spark-worker-controller-1h0l7 -it bash
$ cd /opt/spark

# 提交python spark任务
./bin/spark-submit \
    --executor-memory 4G \
    --master spark://spark-master:7077 \
    examples/src/main/python/wordcount.py \
    "hdfs://hadoop-namenode:9000/caicloud/spark/data"

# 提交scala spark任务
./bin/spark-submit 
    --executor-memory 4G
    --master spark://spark-master:7077
    --class io.caicloud.LinearRegression
    /nfs/caicloud/spark-mllib-1.0-SNAPSHOT.jar
    "hdfs://hadoop-namenode:9000/caicloud/spark/data"

 

2.   经过Zeppelin提交任务

    使用Zeppelin提交时有两种方式：pod exec 和 zeppelin-UI。咱们先建立zeppelin的ReplicationController：

$ kubectl create -f replication-controller/zeppelin-controller.yaml
replicationcontroller "zeppelin-controller"created

查看zeppelin：
$ kubectl get pods -l component=zeppelin
NAME                        READY     STATUS    RESTARTS   AGE
zeppelin-controller-5g25x   1/1       Running   0          5h

 

a) 经过zeppelin exec pods方式提交

$ kubectl exec zeppelin-controller-5g25x -it pyspark

b) 经过zeppelin UI方式提交

使用已建立的Zeppelin pod，设置WebUI的映射端口：

$ kubectl port-forward zeppelin-controller-5g25x 8080:8080

访问 http://localhost:8080/，并提交测试代码：

 

使用Spark的问题

●  Spark的master与worker没法通讯

解决办法：查看Spark Master Service是否正常

●  Spark资源与调度问题

a) worker数量不要超过kubernetes集群的节点数量（若是超过，spark性能会降低）

b) executor-memory不要超过机单台器最大内存，executor-cores不要超过单台机器的CPU核数

c) 若是遇到insufficient resources问题，检查是否有其余任务在运行并kill

d) 同一个kubernetes节点上，可能会分配多个worker，致使性能降低（这时最好从新建立worker）

e) 内存、磁盘IO、网络IO、CPU均可能成为Spark的性能瓶颈

●  任务运行时的问题

SocketTimeoutException: Accept timeout —— spark版本为1.5.2时，试试将jdk版本由1.8下降至1.7

使用Zeppelin的问题

●  Zeppelin pod很大，拉取镜像可能会消耗一段时间，取决于你的网络条件

● 第一次运行Zeppelin时， pipeline可能会花费不少时间（约一分钟），须要比较多的时间来初始化。

● kubectl port-forward可能不会长时间保持稳定状态。若是发现Zeppelin变成断开(disconnected)，port-forward极可能出现故障，这时须要重启

● 从ZeppelinUI提交任务，运行时间不稳定（波动较大）

 

为了测试Spark性能，咱们须要使用大量数据，但kubernetes节点所在的机器磁盘空间有限（20G）。经过为每台机器挂载NFS/GlusterFS大网盘，解决大数据存储的问题。

接下来咱们将介绍，如何使用Spark从NFS中读取数据，以及其中遇到的问题。

1. 首先为每一个kubernetes节点安装nfs-client，挂载nfs网盘（假设NFS-server为10.57.*.33，数据存于/data1T5目录下）。

$ sudo apt-get install nfs-common
$ sudo mkdir -p /data1T5
$ sudo mount -t nfs 10.57.*.33:/data1T5 /data1T5

2. 声明PersistentVolumes(pv)和PersistentVolumeClaims(pvc)，使得Kubernetes能链接NFS数据网盘：

$ kubectl create -f nfs/nfs_pv.yaml
$ kubectl create -f nfs/nfs_pvc.yaml

# nfs/nfs_pv.yaml
apiVersion: v1
kind: PersistentVolume
metadata:
    name: spark-cluster
spec:
    capacity:
        storage: 1535Gi
    accessModes:
        - ReadWriteMany
    nfs:
        server: 10.57.*.33
        path: "/data1T5"

# nfs/nfs_pvc.yaml
apiVersion: v1
kind: PersistentVolumeClaim
metadata:
  name: spark-cluster
spec:
  accessModes:
    - ReadWriteMany
  resources:
    requests:
      storage: 1535Gi

3. 为Spark的ReplicationController增长volumeMounts和volumes两项，并从新create，使得Spark容器能访问NFS数据。

修改spark-master-controller.yaml和spark-worker-controller.yaml，存为nfs/spark-master-controller.nfs.yaml 和 nfs/spark-worker-controller.nfs.yaml

$ kubectl delete rc spark-master-controller
$ kubectl create -f nfs/spark-master-controller.nfs.yaml
$ kubectl delete rc spark-worker-controller
$ kubectl create -f nfs/spark-worker-controller.nfs.yaml

4. 使用exec方式进入master/worker所在的pod，能够看到nfs挂载的/data1T5目录：

$ kubectl exec spark-worker-controller-1h0l7 -it bash
root@spark-worker-controller-1h0l7:/# ls -al /data1T5/

成功挂载后，便可在Spark程序中，使用sc.textFile()读取NFS数据。

注意事项：

a) 若是但愿使用spark从nfs中读取数据，必须事先在kubernetes集群的全部节点上安装nfs-client

b) 使用spark从多台nfs-client读取一台nfs-server数据时，遇到IO瓶颈，读取速率只有30MB~40MB/s