title: Spark系列
第三章 Spark启动与基础使用
3.1 Spark Shell
3.1.1 Spark Shell启动
安装目录的bin目录下面,启动命令:
spark-shell
$SPARK_HOME/bin/spark-shell \
--master spark://hadoop10:7077 \
--executor-memory 512M \
--total-executor-cores 2
注意上面的 cores 参数,若是 0,那么以后这个 spark shell 中运行的代码是不能执行成功的。千万注意。必要要把 cpu cores 和 memory 设置合理。
1、executor memory不能超过虚拟机的内存
2、cpu cores不要超过spark集群能够提供的总cpu cores,否则会使用全部。最好不要使用全部。否则其他程序由于没有cpu core可用,就不能正常运行
参数说明:
--master spark://hadoop10:7077 指定Master的地址
--executor-memory 2G 指定每个worker可用内存为2G
--total-executor-cores 2 指定整个集群使用的cpu核数为2个
注意:
如果启动spark shell时没有指定master地址,但是也可以正常启动spark shell和执行spark shell中的程序,其实是启动了spark的local模式,该模式仅在本机启动一个Driver进程,没有与集群建立联系。
Spark-2.x开始/Spark3.x
Spark Shell 中已经默认将 SparkContext 类初始化为对象 sc
Spark Shell 中已经默认将 SparkSession 类初始化为对象 spark
用户代码如果需要用到,则直接应用 sc,spark 即可
Spark-1.x
Spark Shell 中已经默认将 SparkContext 类初始化为对象 sc
Spark Shell 中已经默认将 SQLContext 类初始化为对象 sqlContext
用户代码如果需要用到,则直接应用sc,sqlContext即可
3.1.2 编写WordCount
在提交WordCount程序之前,先在HDFS集群中的准备一个文件用于做单词统计:
words.txt内容如下:
hello huangbo
hello xuzheng
hello wangbaoqiang
把该文件上传到HDFS文件系统中:
hadoop fs -mkdir -p /spark/wc/input
hadoop fs -put words.txt /spark/wc/input
在 Spark Shell 中提交 WordCount 程序:
sc.textFile("hdfs://hadoop10/spark/wc/input/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop10/spark/wc/output")
查询验证结果:
hadoop fs -ls hdfs://hadoop10/spark/wc/output
hadoop fs -cat hdfs://hadoop10/spark/wc/output/*
说明:
sc
.textFile("hdfs://hadoop10/spark/wc/input/words.txt")
.flatMap(_.split(""))
.map((_,1))
.reduceByKey(_+_)
.saveAsTextFile("hdfs://hadoop10/spark/wc/output")
1、sc
是SparkContext对象,该对象时提交spark程序的入口
2、textFile("hdfs://hadoop10/spark/wc/input/words.txt")是从HDFS中读取数据
底层是通过InputFormat去读取(因为数据在HDFS,从HDFS读取数据的规范就是InputFormat)
RDD[String] (文件中的一行,就是RDD中的一条数据。)
3、flatMap(_.split(" "))
先map,再压平,切割压平
Array(Array("hello", "a"), Array("hello", "b")) =》 Array("hello", "a","hello", "b")
4、map((_,1))
将单词和1构成元组(word,1)
5、reduceByKey(_+_)
按照key进行reduce,并将value累加
6、saveAsTextFile("hdfs://hadoop10/spark/wc/output")
将结果写入到HDFS对应输出目录中
3.2 Spark Submit
利用Spark自带的例子程序执行一个求PI(蒙特卡洛算法)的程序:
[root@hadoop10 bin]# ./spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop10:7077 \
--executor-memory 512m \
--total-executor-cores 2 \
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \
100
运行结果:
理解:
执行通过/software/spark/bin/spark-shell来运行。
如果不设置master那么默认运行在本机节点;
如果设置 --master spark://hadoop10:7077,那么运行在Spark Standalone模式集群。
参数说明:
--class 指定 jar 包的主类
--master 指定jar包提交的模式
详解如下:
1、local
本地模式,本地运行,可以调试(local 1个线程、local[*]不限线程、local[N] N个线程,理想情况下,N设置为你机器的CPU核数)
2、spark
提交到Spark Standalone集群,有Master和Worker进程
3、mesos
将jar包提交到mesos集群,运行在mesos资源管理器框架之上,由mesos负责资源管理,Spark负责任务调度和计算
4、YARN
将jar包提交到yarn集群,运行在yarn资源管理器框架之上,由yarn负责资源管理,Spark负责任务调度和计算
5、cloud
比如AWS的EC2,使用这个模式能很方便的访问Amazon的S3,Spark支持多种分布式存储系统,比如HDFS和S3
--deploy-mode 指定jar的运行方式(默认是 client 模式)
详解如下:
1、client 模式
在提交端生成的JVM会一直等待所有计算过程全部完成才退出,它有两个功能,一个是提交,一个是监控jar包运行(测试环境下使用)
2、cluster 模式
在提交端生成的JVM在提交jar包后会退出,它只有一个提交功能,然后在某一个 worker上会生成一个Driver的JVM,该JVM执行监控jar包运行的功能,等待所有代码运行完毕退出(生产环境下使用 )
application.jar 指定你的jar包的地址
arguments 传递给main()方法的参数
3.2.1 Spark Submit多种运行模式
1、提交任务到本地运行
/software/spark/bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master local[4]\
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1\
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \10
2、提交任务到Spark集群运行
/software/spark/bin/spark-submit \--class org.apache.spark.examples.SparkPi \--master spark://hadoop10:7077 \
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1\
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \10
3、提交Yarn集群,使用Yarn-Client模式
/software/spark/bin/spark-submit \--class org.apache.spark.examples.SparkPi \--masteryarn\
--deploy-mode client \
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1\
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \10
3.3 修改Spark日志级别
3.3.1 临时修改
val sparkContext:SparkContext = new SparkContext(conf)
sparkContext.setLogLevel("WARN")
3.3.2 永久修改
可以通过修改Spark配置文件来Spark日志级别
以下是详细步骤的:
第一步:先进入conf目录
[root@hadoop10 conf]$ cd /software/spark/conf
第二步:准备log4j.properties
[root@hadoop10 conf]$ cp log4j.properties.template log4j.properties
第三步:配置日志级别:
把INFO改成你想要的级别:主要有ERROR, WARN, INFO, DEBUG几种
3.4 Spark的WordCount案例
3.4.0 spark-shell中的WordCount
[root@hadoop10 bin]# cd /software/spark/bin/
[root@hadoop10 bin]# pwd
/software/spark/bin
[root@hadoop10 bin]# spark-shell
2021-11-09 16:57:03,855 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://hadoop10:4040
Spark context available as 'sc' (master = local[*], app id = local-1636448230277).
Spark session available as 'spark'.
Welcome to
____ __
/ __/__ ___ _____/ /__
_\ \/ _ \/ _ `/ __/ '_/
/___/ .__/\_,_/_/ /_/\_\ version 3.1.2
/_/
Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_202)
Type in expressions to have them evaluated.
Type :help for more information.
scala> sc.textFile("file:///home/data/wordcount.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println)
(hadoop,1) (0 + 2) / 2]
(hbase,1)
(hello,3)
(world,1)
scala>
3.4.1 Java7版本WordCount
packagecom.aa.sparkjava.core.wordcount;importorg.apache.spark.SparkConf;importorg.apache.spark.api.java.JavaPairRDD;importorg.apache.spark.api.java.JavaRDD;importorg.apache.spark.api.java.JavaSparkContext;importorg.apache.spark.api.java.function.FlatMapFunction;importorg.apache.spark.api.java.function.Function2;importorg.apache.spark.api.java.function.PairFunction;importorg.apache.spark.api.java.function.VoidFunction;importscala.Tuple2;importjava.util.Arrays;importjava.util.Iterator;/**
* @Author AA
* @Date 2021/11/25 16:46
* @Project bigdatapre
* @Package com.aa.sparkjava.core.wordcount
* WordCountJava java7版本的编写
*/publicclassWordCountJava7{publicstaticvoidmain(String[] args){//一、参数判断if(args.length!=2){System.out.println("Usage:JavaWordCount7: Please enter the necessary parameters:<input><output>");System.exit(1);}//二、编程入口SparkConf conf =newSparkConf();
conf.setMaster("local");
conf.setAppName(WordCountJava7.class.getSimpleName());JavaSparkContext jsc =newJavaSparkContext(conf);//三、读取数据,设置输入路径。JavaRDD<String> lineRDD = jsc.textFile(args[0]);//四、进行逻辑处理,切割压平JavaRDD<String> jrdd1 = lineRDD.flatMap(newFlatMapFunction<String,String>(){@OverridepublicIterator<String>call(String line)throwsException{returnArrays.asList(line.split(" ")).iterator();}});//五、将四中的单词结果和1组合成元组JavaPairRDD<String,Integer> javaPairRDD = jrdd1.mapToPair(newPairFunction<String,String,Integer>(){@OverridepublicTuple2<String,Integer>call(String s)throwsException{returnnewTuple2<String,Integer>(s,1);}});//六、 分组聚合 reduceByKey() (a,b)=>a+b 前两个参数是输入参数类型,第三个参数:返回值的类型JavaPairRDD<String,Integer> result = javaPairRDD.reduceByKey(newFunction2<Integer,Integer,Integer>(){@OverridepublicIntegercall(Integer v1,Integer v2)throwsException{return v1 + v2;}});//七、保存结果或者打印输出//打印输出
result.foreach(newVoidFunction<Tuple2<String,Integer>>(){@Overridepublicvoidcall(Tuple2<String,Integer> tuple)throwsException{System.out.println(tuple);//System.out.println(tuple._1 + " " + tuple._2);}});//保存结果//result.saveAsTextFile(args[1]);//八、释放资源
jsc.close();}}
3.4.2 Java8 Lambda表达式版本WordCount
packagecom.aa.sparkjava.core.wordcount;importorg.apache.spark.SparkConf;importorg.apache.spark.api.java.JavaPairRDD;importorg.apache.spark.api.java.JavaRDD;importorg.apache.spark.api.java.JavaSparkContext;importscala.Tuple2;importjava.util.Arrays;/**
* @Author AA
* @Date 2021/11/26 11:53
* @Project bigdatapre
* @Package com.aa.sparkjava.core.wordcount
* WordCountJava java8版本 使用lambda表达式
*/publicclassWordCountJava8{publicstaticvoidmain(String[] args){if(args.length !=2){System.out.println("Please enter the necessary parameters:Usage Java8 Lambda WordCount<input><output>");System.exit(1);}SparkConf conf =newSparkConf();
conf.setMaster("local");
conf.setAppName(WordCountJava8.class.getSimpleName());JavaSparkContext jsc =newJavaSparkContext(conf);//1、读取数据JavaRDD<String> jrdd = jsc.textFile(args[0]);//2、切割压平JavaRDD<String> jrdd2 = jrdd.flatMap(t ->Arrays.asList(t.split(" ")).iterator());//3、和1组合 JavaPairRDD<String,Integer> jprdd = jrdd2.mapToPair(t ->newTuple2<String,Integer>(t,1));//4、分组聚合JavaPairRDD<String,Integer> res = jprdd.reduceByKey((a, b)-> a + b);//5、保存输出
res.saveAsTextFile(args[1]);//6、释放资源
jsc.close();}}
3.4.3 Scala版本WordCount
packagecom.aa.sparkscala.core.wordcountimportorg.apache.spark.rdd.RDDimportorg.apache.spark.{SparkConf,SparkContext}/**
* @Author AA
* @Date 2021/11/26 15:08
* @Project bigdatapre
* @Package com.aa.sparkscala.core.wordcount
* Spark WordCount Scala版本 使用老的API SparkContext
*/
object WordCountScala{
def main(args:Array[String]):Unit={//1、创建一个SparkConf对象,并设置程序的名称
val conf =newSparkConf().setAppName("WordCount").setMaster("local")//2、创建一个SparkContext对象
val sparkContext:SparkContext=newSparkContext(conf)//3、读取HDFS上的文件构建一个RDD
val fileRDD:RDD[String]= sparkContext.textFile("D:\\input\\test1.txt")//4、构建一个单词RDD
val wordAndOneRDD:RDD[(String,Int)]= fileRDD.flatMap(_.split(" ")).map((_,1))//5、进行单词的聚合
val resultRDD:RDD[(String,Int)]= wordAndOneRDD.reduceByKey(_ + _)//6、保存结果
resultRDD.saveAsTextFile("D:\\output\\wordcountscala1")//7、关闭sc
sparkContext.stop()}}
3.4.4 补充API依赖
<dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.12.14</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.1.2</version></dependency>
声明:
文章中代码及相关语句为自己根据相应理解编写,文章中出现的相关图片为自己实践中的截图和相关技术对应的图片,若有相关异议,请联系删除。感谢。转载请注明出处,感谢。
By luoyepiaoxue2014
B站: https://space.bilibili.com/1523287361 点击打开链接
微博地址: http://weibo.com/luoyepiaoxue2014 点击打开链接
版权归原作者 落叶飘雪2014 所有, 如有侵权,请联系我们删除。