Docker中提交任务到Spark集群

1.  背景描述和需求

数据分析程序部署在Docker中，有一些分析计算需要使用Spark计算，需要把任务提交到Spark集群计算。

接收程序部署在Docker中，主机不在Hadoop集群上。与Spark集群网络互通。

需求如下

1、在Docker中可程序化向Spark集群提交任务

2、在Docker中可对Spark任务管理，状态查询和结束

在Docker中搭建一套Spark、Hadoop环境。任务通过spark-submit --master yarn --deploy-mode cluster来提交到Spark on YARN集群执行。

任务监控通过hadoop的restful接口来监控和管理。

2.1.  Yarn client 模式为行不通

任务发布的docker实例，不在spark集群中，属于非集群机器。只有spark yarn模式的入口，但是hdfs无法与hadoop集群通信。

为什么不能hdfs不能通信？

每个docker启动时，一般不指定ip地址和机器名，不能再集群中预先配置好ip地址和机器名。

并且在hadoop集群中添加了一个动态的docker，但是并不参与任务执行，不利于环境的管理。

在docker中，你的环境配置（python路径，hadoop路径等）可能与hadoop集群不一致，所以以client模式运行时，存在找不到配置的错误。

以cluster模式运行，只要保证把任务所需的文件上传到hadoop集群即可正常运行任务，docker不与集群通信业可以正常执行任务。

Client模式为何不行？

Client模式是由RM分配一个AM，然后由executor反向driver注册，dirver发送task，在回收结果。

但是现在dirver在Docker中，executor找不到dirver的地址，无法注册，所以导致client模式无法使用。

https://www.cnblogs.com/fbiswt/p/4667956.html

1、客户端安装的机器一般是虚拟机，虚拟机的名称可能是随便搞的，然而，yarn-client模式提交任务，是默认把本机当成driver的。所以导致其他的机器无法通过host的name直接访问这台机器。报错就是Failed to connect to driver at x.x.x.x，retrying.....

　解决办法：在命令后面加上一个--conf spark.driver.host=$your_ip_address，后面直接填客户端机器的IP地址就行。还有一个办法：export SPARK_JAVA_OPTS="-Dspark.driver.host=$your_ip_address"，但是这种方法你在用完yarn-client后就没有办法再用yarn-cluster了。千万不能把这个参数配置到spark-default.conf里面。

2.2.  Cluster与client模式异同

Cluster模式，driver节点在集群内部，可以最大限度的减少driver和executor直接的网络延时。

这部分内容有详细介绍。

3.1.  Spark on Yarn集群搭建

资源管理模式设置为YARN模式，开放Hadoop Web管理页面。

详细搭建过程略。网上有详细文档。

3.2.  Web管理页面和rest配置

hadoop目录etc/Hadoop目录yarn-site.xml文件

通过yarn.resourcemanager.webapp.address 设置web访问地址和端口

<!-- Site specific YARN configuration properties -->

 <property>

  <name>yarn.nodemanager.aux-services</name>

  <value>mapreduce_shuffle</value>

</property>

<property>

   <name>yarn.resourcemanager.address</name>

   <value>master:8032</value>

</property>

<property>

  <name>yarn.resourcemanager.scheduler.address</name>

  <value>master:8030</value>

</property>

<property>

 <name>yarn.resourcemanager.resource-tracker.address</name>

 <value>master:8031</value>

</property>

<property>

 <name>yarn.resourcemanager.admin.address</name>

 <value>master:8033</value>

</property>

<property>

 <name>yarn.resourcemanager.webapp.address</name>

 <value>IP address:8088</value>

</property>

4.1.  软件配套表

4.2.  制作Spark的Docker镜像

安装基础软件

把java、spark、hadoop、python的软件放到同一个目录下，在Dockerfile中使用ADD命令添加软件到镜像。

软件放到了./add 目录下，通过ADD命令把目录下的软件添加到Docker中的/目录

软件环境变量设置

这里以把java、spark、hadoop、anaconda3:4.2.0都放到根目录下为例：

Hadoop配置

Hadoop配置文件直接使用spark集群中的hadoop配置即可。路径为hadoop_dir/etc/Hadoop,把目录下的配置都复制过来即可。

提示：

如果Docker中java路径与Spark中一致，则需要修改为docker中的路径。不然docker中spark无法运行

如果提交的是任务是python程序，则PYSPARK_PYTHON设置要求与spark集群的配置保持一致。

通过PYSPARK_PYTHON="/anaconda3/bin/python"来设置

在Docker的程序中，执行用户与spark集群中的执行账户可能不一致，则需要通过环境变量HADOOP_USER_NAME来设置。与集群中保持一致，不然会产生权限问题。

HADOOP_USER_NAME="spark"

Dockerfile示例

FROM your_base_env

ENV LANG="en_US.UTF-8"

 

ADD ./add .

 

# Spark ENV

ENV JAVA_HOME="/jdk1.8.0_121" SPARK_HOME="/spark-2.3.2-bin-hadoop2.7" HADOOP_HOME="/hadoop-2.7.3" CLASSPATH="/jdk1.8.0_121/lib/dt.jar:/jdk1.8.0_121/lib/tools.jar" 

PATH="/jdk1.8.0_121/bin:$PATH:/spark-2.3.2-bin-hadoop2.7/bin:/hadoop-2.7.3/bin" PATH="$PATH:$INSTALL_PATH" PYTHONPATH="$INSTALL_PATH" LANG="en_US.UTF-8" 

HADOOP_CONF_DIR="/hadoop-2.7.3/etc/hadoop/ " PYSPARK_PYTHON="/anaconda3/bin/python" HADOOP_USER_NAME="spark"

4.3.  环境检查

配置完成后，打包镜像，启动一个实例

检查spark配置

输入命令：spark-submit，检查spark是否安装完成

看到下面的信息，说明spark设置成功

[root@3920e4505b70 stg]# spark-submit

Usage: spark-submit [options] <app jar | python file | R file> [app arguments]

Usage: spark-submit --kill [submission ID] --master [spark://...]

Usage: spark-submit --status [submission ID] --master [spark://...]

Usage: spark-submit run-example [options] example-class [example args]

 

Options:

  --master MASTER_URL         spark://host:port, mesos://host:port, yarn,

                              k8s://https://host:port, or local (Default: local[*]).

  --deploy-mode DEPLOY_MODE   Whether to launch the driver program locally ("client") or

                              on one of the worker machines inside the cluster ("cluster")

                              (Default: client).

  --class CLASS_NAME          Your application's main class (for Java / Scala apps).

  --name NAME                 A name of your application.

  --jars JARS                 Comma-separated list of jars to include on the driver

                              and executor classpaths.

  --packages                  Comma-separated list of maven coordinates of jars to include

                              on the driver and executor classpaths. Will search the local

                              maven repo, then maven central and any additional remote

                              repositories given by --repositories. The format for the

                              coordinates should be groupId:artifactId:version.

  --exclude-packages          Comma-separated list of groupId:artifactId, to exclude while

                              resolving the dependencies provided in --packages to avoid

                              dependency conflicts.

检查Hadoop配置

输入命令：yarn application -list 检查hadoop集群配置

看到下面输出表示正常

执行任务Demo

环境检查没有问题，执行demo代码来检查下：

spark-submit --master yarn --deploy-mode cluster /spark-2.3.2-bin-hadoop2.7/examples/src/main/python/pi.py

没有问题，则会看到任务状态为ACCEPTED说明集群接受了任务

RUNNING说明spark集群开始执行任务了。

任务监控的两个方法

• Yarn 命令行
• Hadoop HTTP rest接口

由于开发语言为python，调用shell命令没有rest接口方便，选择使用rest接口来做任务监控方案。

5.1.  yarn application命令监控和管理

通过命令查看当前运行的任务，查看自己运行的任务是否在列表中

yarn application –list 查看任务列表

yarn application –kill  applicationID  结束指定任务

5.2.  hadoop集群的rest接口来管理

在python中调用shell命令，存在诸多不便。

调用rest接口非常方便。这里选择rest接口作为任务状态管理方式。

接口文档地:

https://hadoop.apache.org/docs/r2.7.3/hadoop-yarn/hadoop-yarn-site/ResourceManagerRest.html#Cluster_Application_API

5.2.1.  查询任务列表

GET http://<rm http address:port>/ws/v1/cluster/apps

参数：states=accepted,running,finished

查询条件过滤accepted,running，检查提交的任务是否在任务列表中。

5.2.2.  查询单个任务

如果任务已经接受了，生成了任务id，则可以直接根据任务id来查询任务状态

GET http://<rm http address:port>/ws/v1/cluster/apps/ application_1546828007170_0142

查询刚才任务执行的结果：

{
    "app":{
        "id":"application_1546828007170_0142",
        "user":"csmsopr",
        "name":"pi.py",
        "queue":"default",
        "state":"FINISHED",
        "finalStatus":"SUCCEEDED",
        "progress":100,
        "trackingUI":"History",
        "trackingUrl":"http://host281566:8088/proxy/application_1546828007170_0142/",
        "diagnostics":"",
        "clusterId":1546828007170,
        "applicationType":"SPARK",
        "applicationTags":"",
        "startedTime":1548234101173,
        "finishedTime":1548234115661,
        "elapsedTime":14488,
        "amContainerLogs":"http://host281567:8042/node/containerlogs/container_1546828007170_0142_01_000001/csmsopr",
        "amHostHttpAddress":"host281567:8042",
        "allocatedMB":-1,
        "allocatedVCores":-1,
        "runningContainers":-1,
        "memorySeconds":51782,
        "vcoreSeconds":32,
        "preemptedResourceMB":0,
        "preemptedResourceVCores":0,
        "numNonAMContainerPreempted":0,
        "numAMContainerPreempted":0
    }
}

5.2.3.  任务状态管理

任务状态查询和结束

返回

查询任务状态

返回：