1. 程式人生 > >3.2、spark集群運行應用之第三方jar的處理方式

3.2、spark集群運行應用之第三方jar的處理方式

print ext collect each bsp com converter 應用 ins

在編寫程序時,不可避免會用到第三方jar,有三種使用方式:

1、將運行程序需要的所有第三方 jar,分發到所有spark的/soft/spark/jars下

2、將第三方jar打散,和自己的源碼打成一個jar包,如3.1中

3、在spark-submit命令中,通過--jars指定使用的第三方jar包

  在s102上提交,fastjson-1.2.47.jar 本地,myspark.jar本地,temptags.txt HDFS上

spark-submit --class a --jars fastjson-1.2.47.jar --master spark://s101:7077 myspark.jar temptags.txt

  spark-shell腳本也用到spark-submit,因此也可以通過spark-shell指定第三方 jar

spark-shell --master spark://s101:7077 --jars fastjson-1.2.47.jar  //該jar在本地
import java.util._
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import com.alibaba.fastjson._
def pp(line: String)={  //解析方法
    val list 
= new ArrayList[String] val jsonObject = JSON.parseObject(line) val extInfoList = jsonObject.getJSONArray("extInfoList") if (extInfoList != null && extInfoList.size != 0) { for (o <- extInfoList) { val jo = o.asInstanceOf[JSONObject] if (jo.get
("title") == "contentTags") { val values = jo.getJSONArray("values") for (value <- values) { list.add(value.toString) } } } } list } val rdd1 = sc.textFile("myspark/temptags.txt") val rdd2 = rdd1.map(s => {val sp = s.split("\t");val lst = pp(sp(1));(sp(0), lst)}).filter(_._2.size() > 0) val rdd3 = rdd2.flatMapValues(_.asScala).map(t=>((t._1,t._2),1)).reduceByKey((a,b)=>a+b).groupBy(_._1._1).mapValues(_.map(t=>(t._1._2,t._2))) val rdd4 = rdd3.mapValues(_.toList.sortBy(-_._2)).sortBy(-_._2(0)._2) val rdd5 = rdd4.collect() rdd5.foreach(println)

3.2、spark集群運行應用之第三方jar的處理方式