13518219792
大数据学习路线分享Master的jps，SparkSubmit
成都创新互联公司是一家以重庆网站建设公司、网页设计、品牌设计、软件运维、营销推广、小程序App开发等移动开发为一体互联网公司。已累计为户外休闲椅等众行业中小客户提供优质的互联网建站和软件开发服务。
类启动后的服务进程,用于提交任务,
哪一段启动提交任务,哪一段启动submit(Driver端)
提交任务流程
1.Driver端提交任务到Master(启动sparkSubmit进程)
2.Master生成任务信息,放入对列中
3.Master通知Worker启动Executor,(Master过滤出存活的Worker,将任务分配给空闲资源多的worker)
4.worker的Executor向Driver端注册(只有executor真正参与计算) -> worker从Dirver端拿信息
5.Driver端启动Executor将任务划分阶段,分成小的task,再广播给相应的Worker让他去执行
6.worker会将执行完的任务回传给Driver
range 相当于集合子类
scala> 1.to(10) res0: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
scala> 1 to 10 res1: scala.collection.immutable.Range.Inclusive = Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) |
提交任务到集群的任务类 :
Spark context available as sc
SQL context available as sqlContext
直接调用:
spark WordCount
构建模板代码:
SparkConf：构建配置信息类，该配置优先于集群配置文件
setAppName：指定应用程序名称，如果不指定，会自动生成一个类似于uuid产生的名称
setMaster：指定运行模式：local-用1个线程模拟集群运行，
local[2]: 用2个线程模拟集群运行，loca[*]-当前有多少空闲到的线程就用多少线程来运行该任务
/** * 用spark实现单词计数 */ object SparkWordCount { def main(args: Array[String]): Unit = { /** * 构建模板代码 */ val conf: SparkConf = new SparkConf() .setAppName("SparkWordCount") // .setMaster("local[2]")
// 创建提交任务到集群的入口类（上下文对象） val sc: SparkContext = new SparkContext(conf)
// 获取HDFS的数据 val lines: RDD[String] = sc.textFile(args(0))
// 切分数据，生成一个个单词 val words: RDD[String] = lines.flatMap(_.split(" "))
// 把单词生成一个个元组 val tuples: RDD[(String, Int)] = words.map((_, 1))
// 进行聚合操作 // tuples.reduceByKey((x, y) => x + y) val sumed: RDD[(String, Int)] = tuples.reduceByKey(_+_)
// 以单词出现的次数进行降序排序 val sorted: RDD[(String, Int)] = sumed.sortBy(_._2, false)
// 打印到控制台 // println(sorted.collect.toBuffer) // sorted.foreach(x => println(x)) // sorted.foreach(println)
// 把结果存储到HDFS sorted.saveAsTextFile(args(1))
// 释放资源 sc.stop() } } |
打包后上传Linux
1.首先启动zookeeper,hdfs和Spark集群
启动hdfs
/usr/local/hadoop-2.6.1/sbin/start-dfs.sh
启动spark
/usr/local/spark-1.6.1-bin-hadoop2.6/sbin/start-all.sh
2.使用spark-submit命令提交Spark应用（注意参数的顺序）
/usr/local/spark-1.6.1-bin-hadoop2.6/bin/spark-submit \
--class com.qf.spark.WordCount \
--master spark://node01:7077 \
--executor-memory 2G \
--total-executor-cores 4 \
/root/spark-mvn-1.0-SNAPSHOT.jar \
hdfs://node01:9000/words.txt \
hdfs://node01:9000/out
3.查看程序执行结果
hdfs dfs -cat hdfs://node01:9000/out/part-00000
javaSparkWC
import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.FlatMapFunction; import org.apache.spark.api.java.function.Function2; import org.apache.spark.api.java.function.PairFunction; import scala.Tuple2;
import java.util.Arrays; import java.util.List;
public class JavaSparkWC { public static void main(String[] args) { SparkConf conf = new SparkConf() .setAppName("JavaSparkWC").setMaster("local[1]");
//提交任务入口类 JavaSparkContext jsc = new JavaSparkContext(conf);
//获取数据 JavaRDD //切分数据 JavaRDD lines.flatMap(new FlatMapFunction @Override public Iterable List return splited; } });
//生成元祖 //一对一组 ,(输入单词,输出单词,输出1) JavaPairRDD words.mapToPair(new PairFunction @Override public Tuple2 return new Tuple2 } });
//聚合 //2个相同key的value,聚合 JavaPairRDD tuples.reduceByKey(new Function2 @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } });
//此前key为String类型,没有办法排序 //Java api并没有提供sortBy算子，此时需要把两个值位置调换，排序完成后，在换回来 final JavaPairRDD sumed.mapToPair(new PairFunction @Override public Tuple2 // return new Tuple2 return tup.swap(); //swap(),交换方法 } });
//降序排序 JavaPairRDD //再次交换 JavaPairRDD new PairFunction @Override public Tuple2 return tup.swap(); } });
System.out.println(res.collect());
jsc.stop();//释放资源 } } |
