|
SparkSubmit 类启动后的服务进程,用于提交任务, 哪一段启动提交任务,哪一段启动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<String> lines = jsc.textFile("hdfs://hadoop01:9000/wordcount/input/a.txt"); //切分数据 JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterable<String> call(String s) throws Exception { List<String> splited = Arrays.asList(s.split(" ")); //生成list return splited; } }); //生成元祖 //一对一组 ,(输入单词,输出单词,输出1) JavaPairRDD<String, Integer> tuples = words.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String s) throws Exception { return new Tuple2<String, Integer>(s, 1); } }); //聚合 //2个相同key的value,聚合 JavaPairRDD<String, Integer> sumed = tuples.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); //此前key为String类型,没有办法排序 //Java api并没有提供sortBy算子，此时需要把两个值位置调换，排序完成后，在换回来 final JavaPairRDD<Integer, String> swaped = sumed.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() { @Override public Tuple2<Integer, String> call(Tuple2<String, Integer> tup) throws Exception { // return new Tuple2<Integer, String>(tup._2, tup._1); return tup.swap(); //swap(),交换方法 } }); //降序排序 JavaPairRDD<Integer, String> sorted = swaped.sortByKey(false); //再次交换 JavaPairRDD<String, Integer> res = sorted.mapToPair( new PairFunction<Tuple2<Integer, String>, String, Integer>() { @Override public Tuple2<String, Integer> call(Tuple2<Integer, String> tup)throws Exception { return tup.swap(); } }); System.out.println(res.collect()); jsc.stop();//释放资源 } } |
|
|
