怎么做全屏网站,网站设计模板怎么使用,怎么建立视频网站,内部链接网站大全众所周知#xff0c;Spark是一个快速、通用的大规模数据处理平台#xff0c;和Hadoop的MapReduce计算框架类似。但是相对于MapReduce#xff0c;Spark凭借其可伸缩、基于内存计算等特点#xff0c;以及可以直接读写Hadoop上任何格式数据的优势#xff0c;使批处理更加高效…众所周知Spark是一个快速、通用的大规模数据处理平台和Hadoop的MapReduce计算框架类似。但是相对于MapReduceSpark凭借其可伸缩、基于内存计算等特点以及可以直接读写Hadoop上任何格式数据的优势使批处理更加高效并有更低的延迟。实际上Spark已经成为轻量级大数据快速处理的统一平台。 Spark作为一个数据计算平台和框架更多的是关注Spark Application的管理而底层实际的资源调度和管理更多的是依靠外部平台的支持 Spark官方支持四种Cluster ManagerSpark standalone cluster manager、Mesos、YARN和Kubernetes。由于我们TalkingData是使用Kubernetes作为资源的调度和管理平台所以Spark On Kubernetes对于我们是最好的解决方案。
如何搭建生产可用的Kubernetes集群
部署
目前市面上有很多搭建Kubernetes的方法比如Scratch、Kubeadm、Minikube或者各种托管方案。因为我们需要简单快速地搭建功能验证集群所以选择了Kubeadm作为集群的部署工具。部署步骤很简单在master上执行
kubeadm init
在node上执行
kubeadm join --token : --discovery-token-ca-cert-hash sha256:
具体配置可见官方文档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的配置示例如下
{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实际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的网络转发
-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访问其他网段的话就通过物理网关10.0.127.254。 还有一个需要注意的地方是出于kernel security的考虑link物理接口的macvlan是无法直接和物理接口通信的这就导致容器并不能将宿主机作为网关。我们采用了一个小技巧避开了这个限制。我们从物理接口又创建了一个macvlan将物理IP移到了这个接口上物理接口只作为网络入口
$ cat /etc/sysconfig/network-scripts/ifcfg-eth0
DEVICEeth0
IPV6INITno
BOOTPROTOnone
$ cat /etc/sysconfig/network-scripts/ifcfg-macvlan
DEVICEmacvlan
NAMEmacvlan
BOOTPROTOnone
ONBOOTyes
TYPEmacvlan
DEVICETYPEmacvlan
DEFROUTEyes
PEERDNSyes
PEERROUTESyes
IPV4_FAILURE_FATALno
IPADDR10.0.0.61
PREFIX17
GATEWAY10.0.127.254
MACVLAN_PARENTeth0
MACVLAN_MODEbridge
这样两个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时认为第二个包是重复的因为有相同的五元组从而丢包。具体可看我提的issuehttps://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设置hostNetworktrue使用宿主机网络提供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 driverDriver调用Kubernetes API创建executor podsexecutor pods执行作业代码计算作业结束executor pods回收并清理driver pod处于completed状态保留日志直到Kubernetes GC或者手动清理
先决条件
Spark 2.3Kubernetes 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可以通过以下命令构建镜像
$ ./bin/docker-image-tool.sh -r repo -t my-tag build
$ ./bin/docker-image-tool.sh -r repo -t my-tag push
提交作业
在构建Spark镜像后我们可以通过以下命令提交作业
$ 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.instances5 \--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的作业代码和依赖我们可以在--jars、--files和最后位置指定协议支持http、https和HDFS。 执行提交命令后会有以下输出 任务结束会输出 访问Spark Driver UI
我们可以在本地使用kubectl port-forward访问Driver UI
$ kubectl port-forward driver-pod-name 4040:4040
执行完后通过http://localhost:4040访问。
访问日志
Spark的所有日志都可以通过Kubernetes API和kubectl CLI进行访问
$ kubectl -nnamespace logs -f driver-pod-name
如何实现租户和资源隔离
Kubernetes Namespace
在Kubernetes中我们可以使用namespace在多用户间实现资源分配、隔离和配额。Spark On Kubernetes同样支持配置namespace创建Spark作业。 首先创建一个Kubernetes namespace
$ kubectl create namespace spark
由于我们的Kubernetes集群使用了RBAC所以还需创建serviceaccount和绑定role
$ kubectl create serviceaccount spark -n spark
$ kubectl create clusterrolebinding spark-role --clusterroleedit --serviceaccountspark:spark --namespacespark
并在spark-submit中新增以下配置
$ bin/spark-submit \--conf spark.kubernetes.authenticate.driver.serviceAccountNamespark \--conf spark.kubernetes.namespacespark \...
资源隔离
考虑到我们Spark作业的一些特点和计算资源隔离前期我们还是选择了较稳妥的物理隔离方案。具体做法是为每个组提供单独的Kubernetes namespace计算任务都在各自namespace里提交。计算资源以物理机为单位折算成cpu和内存纳入Kubernetes统一管理。在Kubernetes集群里通过node label和PodNodeSelector将计算资源和namespace关联。从而实现在提交Spark作业时计算资源总是选择namespace关联的node。 具体做法如下 1、创建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: nulllabels:component: kube-apiservertier: control-planename: kube-apiservernamespace: kube-system
spec:containers:- command:- kube-apiserver- --secure-port6443- --proxy-client-cert-file/etc/kubernetes/pki/front-proxy-client.crt- --admission-controlInitializers,NamespaceLifecycle,LimitRanger,ServiceAccount,DefaultStorageClass,DefaultTolerationSeconds,NodeRestriction,ResourceQuota,MutatingAdmissionWebhook,ValidatingAdmissionWebhook,PodNodeSelector
...
3、配置PodNodeSelector 在namespace的annotations中添加scheduler.alpha.kubernetes.io/node-selector: sparkspark。
apiVersion: v1
kind: Namespace
metadata:annotations:scheduler.alpha.kubernetes.io/node-selector: sparksparkname: spark
完成以上配置后可以通过spark-submit测试结果
$ spark-submit
--conf spark.kubernetes.authenticate.driver.serviceAccountNamespark
--conf spark.kubernetes.namespacespark
--master k8s://https://xxxx:6443
--deploy-mode cluster
--name spark-pi
--class org.apache.spark.examples.SparkPi
--conf spark.executor.instances5
--conf spark.kubernetes.container.imagexxxx/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作业全分配到了关联的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的日志
$ kubectl -nnamespace logs -f driver-pod-name
收集和聚合日志我们后面会和ES结合。 监控 我们TalkingData内部有自己的监控平台OWL[2]已开源未来我们计划编写metric plugin将Kubernetes接入OWL中。 混合部署 为了保证Spark作业时刻有可用的计算资源我们前期采用了物理隔离的方案。显而易见这种方式大幅降低了物理资源的使用率。下一步我们计划采用混部方案通过以下三种方式实现
将HDFS和Kubernetes混合部署为Spark作业和Kubernetes node划分优先级在低优先级的node上同时运行一些无状态的其他生产服务利用云实现资源水平扩展以防止资源突增
资源扩展
在采用以下两种方法增加资源使用率时集群可能会面临资源短缺和可用性的问题
混合部署资源超卖
这会导致运行资源大于实际物理资源的情况我称之为资源挤兑。一种做法是给资源划分等级优先保证部分等级的资源供给。另一种做法是实现资源的水平扩展动态补充可用资源并在峰值过后自动释放。
原文链接 本文为云栖社区原创内容未经允许不得转载。
