SparkContext初始化
相关知识
- SparkConf 是
SparkContext
的构造参数,储存着Spark
相关的配置信息,且必须指定Master
(比如Local
)和AppName
(应用名称),否则会抛出异常; - SparkContext 是程序执行的入口,一个
SparkContext
代表一个Application
。
初始化过程的主要核心:
- 依据
SparkConf
创建一个Spark
执行环境SparkEnv
; - 创建并初始化
Spark UI
,方便用户监控,默认端口为4040
; - 设置
Hadoop
相关配置及Executor
环境变量; - 创建和启动
TaskScheduler
,DAGScheduler
。
初始化方式:
SparkConf conf = new SparkConf().setAppName(appName).setMaster(master)
JavaSparkContext sc=new JavaSparkContext(conf)
程序运行完后需使用sc.stop()
关闭SparkContext
。
import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.SparkConf; import java.util.Arrays; import java.util.List;public class Edu {public static void main(String[] args) {/********** Begin **********///第一步:设置SparkConfSparkConf conf = new SparkConf().setAppName("educoder").setMaster("local");//第二步:初始化SparkContextJavaSparkContext sc = new JavaSparkContext(conf);/********** End **********/List<String> data = Arrays.asList("hello");JavaRDD<String> r1 = sc.parallelize(data);System.out.print(r1.collect());/********** Begin **********///第三步:关闭SparkContextsc.stop();/********** End **********/} }
集合并行化创建RDD
任务描述
本关任务:计算并输出各个学生的总成绩。
相关知识
为了完成本关任务,你需要掌握:1.集合并行化创建RDD
,2.reduceByKey
。
集合创建RDD
Spark
会将集合中的数据拷贝到集群上去,形成一个分布式的数据集合,也就是一个RDD
。相当于是,集合中的部分数据会到一个节点上,而另一部分数据会到其他节点上。然后就可以用并行的方式来操作这个分布式数据集合,即RDD
。
ArrayList<Integer> list = new ArrayList<Integer>();
list.add(1);
list.add(2);
list.add(3);
JavaRDD<Integer> rdd = sc.parallelize(list,3);//参数1:Seq集合,必须。参数2:分区数,默认为该Application分配到的资源的CPU核数
Integer sum = rdd.reduce((a, b) -> a + b);
System.out.print(sum);
输出:6
reduceByKey()
对元素为RDD[K,V]
对的RDD
中Key
相同的元素的Value
进行聚合。
List<Tuple2<String,Integer>> list = Arrays.asList(new Tuple2("hive",2),new Tuple2("spark",4),new Tuple2("hive",1));
JavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list);
List<Tuple2<String, Integer>> result = listRDD.reduceByKey((x, y) -> x + y).collect();
输出: (spark,4)
(hive,3)
collect() :以数组的形式返回RDD
中的所有元素,收集分布在各个worker
的数据到driver
节点。
编程要求
根据提示,在右侧编辑器begin-end
处补充代码,计算并输出各个学生的总成绩。
("bj",88)
:bj
指学生姓名,88
指学生成绩。
测试说明
平台会对你编写的代码进行测试:
预期输出: (bj,254)
(sh,221)
(gz,285)
package step1;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;
import java.util.*;
public class JStudent {public static void main(String[] args) {SparkConf conf = new SparkConf().setMaster("local").setAppName("JStudent");JavaSparkContext sc = new JavaSparkContext(conf);List<Tuple2<String,Integer>> list = Arrays.asList(new Tuple2("bj",88),new Tuple2("sh",67),new Tuple2("gz",92),new Tuple2("bj",94),new Tuple2("sh",85),new Tuple2("gz",95),new Tuple2("bj",72),new Tuple2("sh",69),new Tuple2("gz",98));//第一步:创建RDDJavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list);//第二步:把相同key的进行聚合JavaPairRDD<String, Integer> result = listRDD.reduceByKey((x, y) -> x + y);//第三步:收集List<Tuple2<String, Integer>> collect = result.collect();//第四步:输出for (Tuple2 c:collect){System.out.println(c);}sc.stop();}
}
读取外部数据集创建RDD
任务描述
本关任务:读取文本文件,按照文本中数据,输出老师及其出现次数。
相关知识
为了完成本关任务,你需要掌握:1.读取文件创建RDD
,2.本关所需算子。
读取文件
textFile()
JavaRDD<String> rdd = sc.textFile("/home/student.txt")//文件路径
算子
(1)mapToPair:此函数会对一个RDD
中的每个元素调用f
函数,其中原来RDD
中的每一个元素都是T
类型的,调用f
函数后会进行一定的操作把每个元素都转换成一个<K2,V2>
类型的对象
ArrayList<Integer> list = new ArrayList<Integer>();
list.add(1);
list.add(2);
list.add(3);
JavaRDD<Integer> rdd = sc.parallelize(list);
JavaPairRDD<Integer,String> result = rdd.mapToPair(x -> new Tuple2(x,1)
输出:(1,1)(2,1)(3,1)
(2) reduceByKey() :对元素为RDD[K,V]
对的RDD
中Key
相同的元素的Value
进行聚合
List<Tuple2<String,Integer>> list = Arrays.asList(new Tuple2("hive",2),new Tuple2("spark",4),new Tuple2("hive",1));
JavaPairRDD<String, Integer> listRDD = sc.parallelizePairs(list);
List<Tuple2<String, Integer>> result = listRDD.reduceByKey((x, y) -> x + y).collect();
输出: (spark,5)
(hive,3)
编程要求
根据提示,在右侧编辑器begin-end
处补充代码,输出老师姓名和出现次数。
- 输入文件样例:
bigdata,laozhang
bigdata,laoduan
javaee,xiaoxu
bigdata
指科目,laozhang
指老师名称。
预期输出: (laoliu,1)
(laoli,3)
(laoduan,5)
(laozhang,2)
(laozhao,15)
(laoyang,9)
(xiaoxu,4)
package step2;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 scala.Tuple2;import java.util.Arrays;
import java.util.List;public class JTeachers {public static void main(String[] args) {SparkConf conf = new SparkConf().setMaster("local").setAppName("JTeachers");JavaSparkContext sc = new JavaSparkContext(conf);String dataFile = "file:///root/step2_files";//第一步:以外部文件方式创建RDDJavaRDD<String> teaRDD = sc.textFile(dataFile);//String name = line.split(",")[1];//第二步:将文件中每行的数据切分,得到自己想要的返回值Integer one = 1;JavaPairRDD<String, Integer> teacher = teaRDD.mapToPair(line ->{String names = line.split(",")[1];Tuple2<String, Integer> t2 = new Tuple2<>(names, one);return t2;});//第三步:将相同的key进行聚合JavaPairRDD<String, Integer> tea = teacher.reduceByKey((x, y) -> x + y);//第四步:将结果收集起来List<Tuple2<String, Integer>> result = tea.collect();//第五步:输出for (Tuple2 t:result){System.out.println(t);}sc.stop();}
}
map
算子完成转换操作
相关知识
为了完成本关任务,你需要掌握:如何使用map
算子。
map
将原来RDD
的每个数据项通过map
中的用户自定义函数f
映射转变为一个新的元素。
图中每个方框表示一个RDD
分区,左侧的分区经过自定义函数f:T->U
映射为右侧的新RDD
分区。但是,实际只有等到Action
算子触发后,这个f
函数才会和其他函数在一个Stage
中对数据进行运算。
map 案例
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6);
System.out.println("init:" + list);
JavaRDD<Integer> rdd = sc.parallelize(list);
JavaRDD<Integer> map = rdd.map(x -> x * 2);
System.out.println("result :" + map.collect());
输出:
init :[1, 2, 3, 4, 5, 6]
result :[2, 4, 6, 8, 10, 12]
说明:rdd1
的元素(1 , 2 , 3 , 4 , 5 , 6
)经过map
算子(x -> x*2
)转换成了rdd2
(2 , 4 , 6 , 8 , 10
)
编程要求
根据提示,在右侧编辑器begin-end
处补充代码,完成以下需求:
需求1:使用map
算子,将rdd
的数据(1, 2, 3, 4, 5)
按照下面的规则进行转换操作,规则如下:
-
偶数转换成该数的平方;
-
奇数转换成该数的立方。
需求2:使用map
算子,将rdd
的数据("dog", "salmon", "salmon", "rat", "elephant")
按照下面的规则进行转换操作,规则如下:
- 将字符串与该字符串的长度组合成一个元组,例如
dog --> (dog,3)
salmon --> (salmon,6)
package net.educoder; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import scala.Tuple2; import java.util.Arrays; import java.util.List; public class Step1 {private static SparkConf conf;private static JavaSparkContext sc;static {conf = new SparkConf().setAppName("Step1").setMaster("local");sc = new JavaSparkContext(conf);}/*** 返回JavaRDD** @return JavaRDD*/public static JavaRDD<Integer> MapRdd() {List<Integer> list = Arrays.asList(1, 2, 3, 4, 5);JavaRDD<Integer> rdd = sc.parallelize(list);/**** 需求:使用map算子,将rdd的数据进行转换操作* 规则如下:* 偶数转换成该数的平方* 奇数转换成该数的立方**//********** begin ***********/JavaRDD<Integer> map = rdd.map(num -> {if (num % 2 == 0) {return num * num;} else {return num * num * num;}});return map;/********** end ***********/}/*** 返回JavaRDD** @return JavaRDD*/public static JavaRDD<Tuple2> MapRdd2() {List<String> list = Arrays.asList("dog", "salmon", "salmon", "rat", "elephant");JavaRDD<String> rdd = sc.parallelize(list);/**** 需求:使用map算子,将rdd的数据进行转换操作* 规则如下:* 将字符串与该字符串的长度组合成一个元组,例如:dog --> (dog,3),salmon --> (salmon,6)**//********** begin ***********/JavaRDD<Tuple2> map = rdd.map(str -> {int i = str.length();return new Tuple2(str, i);});return map;/********** end ***********/} }