如何停止提交到spark集群上的任务

这里是结合Hadoop2.0使用的1，download:根据下载的spark的README中的描述下载合适的版本3，安装其实就是解压，配置/etc/profile环境变量exportSPARK_HOME=/data1/spark/sparkexportSCALA_HOME=/data1/spark/scala-2.9.3exportPATH=$PATH:$SPARK_HOME/bin:$SCALA_HOME/bin配置spark的conf下的spark-env.shexportJAVA_HOME=/usr/java/defaultexportSCALA_HOME=/data1/spark/scala-2.9.3exportSPARK_MASTER_IP=192.168.0.1exportSPARK_MASTER_WEBUI_PORT=8080exportSPARK_WORKER_WEBUI_PORT=8000exportYARN_CONF_DIR=/data/hadoop/hadoop-2.0/etc/hadoop配置slaves(ip根据需要修改)192.168.0.2192.168.0.3分发spark目录和scala目录到几台服务器相同路径下4,启动进入主节点的spark目录的bin下stop-all.sh是停掉集群，start-all.sh启动集群，jps可以在主节点看到master进程，slave节点看到worker进程5，运行程序，运行例子进入spark目录下分布式运行./run-exampleorg.apache.spark.examples.SparkPispark://192.168.0.1:7077./run-exampleorg.apache.spark.examples.SparkLRspark://192.168.0.1:7077本地运行./run-exampleorg.apache.spark.examples.SparkPilocal./run-exampleorg.apache.spark.examples.SparkLRlocal

我们在windows开发机上使用spark的local模式读取远程hadoop集群中的hdfs上的数据，这样的目的是方便快速调试，而不用每写一行代码或者一个方法，一个类文件都需要打包成jar上传到linux上，再扔到正式的集群上进行测试，像功能性验证直接使用local模式来快速调测是非常方便的，当然功能测试之后，我们还需要打包成jar仍到集群上进行其他的验证比如jar包的依赖问题，这个在local模式是没法测的，还有集群运行的调优参数，这些都可以在正式仍到集群时验证。

一个样例代码如下：

def main(args: Array[String]): Unit = {//指定local模式

val conf = new SparkConf().setMaster("local[2]").setAppName("read kp data to kafka")val sc= new SparkContext(conf)//支持通配符路径，支持压缩文件读取

val rrd=sc.textFile("hdfs://192.168.10.4:8020/data/log/{20170227,20170228}/tomcat-log*")//提到到集群模式时，去掉uri地址，如果有双namenode，可以自动容灾

//val rrd=sc.textFile("/data/log/{20170227,20170228}/tomcat-log*")

//统计数量

println(rrd.count())//停止spark

sc.stop()

}

如何在spark中遍历数据时获取文件路径：

val path:String="hdfs://192.168.10.4:8020/data/userlog/{20170226}/kp*"

val text= sc.newAPIHadoopFile[LongWritable,Text,TextInputFormat](path)

val linesWithFileNames = text.asInstanceOf[NewHadoopRDD[LongWritable, Text]]

.mapPartitionsWithInputSplit((inputSplit, iterator) =>{

val file = inputSplit.asInstanceOf[FileSplit]iterator.map(tup =>(file.getPath, tup._2)) // 返回的K=全路径 V=每一行的值

}

)

linesWithFileNames.foreach(println)

如果遍历压缩文件时想要获取文件名，就使用newAPIHadoopFile，此外在本地调试下通过之后，提交到集群运行的时候，一定要把uri去掉，本地加上是想让它远程读取方便调试使用，如果正式运行去掉uri在双namenode的时候可以自动兼容，不去反而成一个隐患了。

最后我们可以通过spark on yarn模式提交任务，一个例子如下：

jars=`echo /home/search/x_spark_job/libs/*jar | sed 's/ /,/g'`

bin/spark-submit --class KSearch --master yarn --jars $jars/home/search/x_spark_job/kp-1.0.0.jar

这里用spark提交有另外一个优势，就是假如我开发的不是YARN应用，就是代码里没有使用SparkContext，而是一个普通的应用，就是读取mysql一个表的数据，写入另外一个mysql，这里跟MR没有关系，但是我依然可以用spark-sumbit提交，这时候是不会提交到YARN上的，但是程序会按普通程序运行，程序依赖的jar包，直接使用--jars传入就行，这一点非常方便，尤其是应用有多个依赖时，比如依赖es，hadoop，hbase，redis，fastjson，我打完包后的程序是瘦身的只有主体jar非常小，依赖的jar我可以不打到主体jar里面，在外部用的时候传入，方便共用并灵活性大大提高。

---