title: Spark系列
安装目录的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即可
在提交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对应输出目录中
利用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()方法的参数
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 \
--master yarn \
--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
val sparkContext:SparkContext = new SparkContext(conf)
sparkContext.setLogLevel("WARN")
可以通过修改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几种
[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>
package com.aa.sparkjava.core.wordcount;
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 org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;
import java.util.Arrays;
import java.util.Iterator;
/**
* @Author AA
* @Date 2021/11/25 16:46
* @Project bigdatapre
* @Package com.aa.sparkjava.core.wordcount
* WordCountJava java7版本的编写
*/
public class WordCountJava7 {
public static void main(String[] args){
//一、参数判断
if(args.length!=2){
System.out.println("Usage:JavaWordCount7: Please enter the necessary parameters:<input><output>");
System.exit(1);
}
//二、编程入口
SparkConf conf = new SparkConf();
conf.setMaster("local");
conf.setAppName(WordCountJava7.class.getSimpleName());
JavaSparkContext jsc = new JavaSparkContext(conf);
//三、读取数据,设置输入路径。
JavaRDD<String> lineRDD = jsc.textFile(args[0]);
//四、进行逻辑处理,切割压平
JavaRDD<String> jrdd1 = lineRDD.flatMap(new FlatMapFunction<String, String>() {
@Override
public Iterator<String> call(String line) throws Exception {
return Arrays.asList(line.split(" ")).iterator();
}
});
//五、将四中的单词结果和1组合成元组
JavaPairRDD<String, Integer> javaPairRDD = jrdd1.mapToPair(new PairFunction<String, String, Integer>() {
@Override
public Tuple2<String, Integer> call(String s) throws Exception {
return new Tuple2<String, Integer>(s, 1);
}
});
//六、 分组聚合 reduceByKey() (a,b)=>a+b 前两个参数是输入参数类型,第三个参数:返回值的类型
JavaPairRDD<String, Integer> result = javaPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
//七、保存结果或者打印输出
//打印输出
result.foreach(new VoidFunction<Tuple2<String, Integer>>() {
@Override
public void call(Tuple2<String, Integer> tuple) throws Exception {
System.out.println(tuple);
//System.out.println(tuple._1 + " " + tuple._2);
}
});
//保存结果
//result.saveAsTextFile(args[1]);
//八、释放资源
jsc.close();
}
}
package com.aa.sparkjava.core.wordcount;
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;
/**
* @Author AA
* @Date 2021/11/26 11:53
* @Project bigdatapre
* @Package com.aa.sparkjava.core.wordcount
* WordCountJava java8版本 使用lambda表达式
*/
public class WordCountJava8 {
public static void main(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 = new SparkConf();
conf.setMaster("local");
conf.setAppName(WordCountJava8.class.getSimpleName());
JavaSparkContext jsc = new JavaSparkContext(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 -> new Tuple2<String, Integer>(t, 1));
//4、分组聚合
JavaPairRDD<String, Integer> res = jprdd.reduceByKey((a, b) -> a + b);
//5、保存输出
res.saveAsTextFile(args[1]);
//6、释放资源
jsc.close();
}
}
package com.aa.sparkscala.core.wordcount
import org.apache.spark.rdd.RDD
import org.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 = new SparkConf().setAppName("WordCount").setMaster("local")
//2、创建一个SparkContext对象
val sparkContext: SparkContext = new SparkContext(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()
}
}
<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>
声明:
文章中代码及相关语句为自己根据相应理解编写,文章中出现的相关图片为自己实践中的截图和相关技术对应的图片,若有相关异议,请联系删除。感谢。转载请注明出处,感谢。
B站: https://space.bilibili.com/1523287361 点击打开链接
微博地址: http://weibo.com/luoyepiaoxue2014 点击打开链接
文章浏览阅读3.2w次,点赞16次,收藏90次。对于这个问题我也是从网上找了很久,终于解决了这个问题。首先遇到这个问题,应该确认虚拟机能不能正常的上网,就需要ping 网关,如果能ping通说明能正常上网,不过首先要用命令route -n来查看自己的网关,如下图:第一行就是默认网关。现在用命令ping 192.168.1.1来看一下结果:然后可以看一下电脑上面百度的ip是多少可以在linux里面ping 这个IP,结果如下:..._linux桥接ping不通baidu
文章浏览阅读512次。小妹在这里已经卡了2-3天了,研究了很多人的文章,除了低版本api 17有成功外,其他的不是channel null 就是没反应 (channel null已解决)拜托各位大大,帮小妹一下,以下是我的程式跟 gradle, 我在这里卡好久又没有人可问(哭)![image](/img/bVcL0Qo)public class MainActivity extends AppCompatActivit..._android 权限申请弹窗 横屏
文章浏览阅读1.4k次,点赞4次,收藏6次。valid padding(有效填充):完全不使用填充。half/same padding(半填充/相同填充):保证输入和输出的feature map尺寸相同。full padding(全填充):在卷积操作过程中,每个像素在每个方向上被访问的次数相同。arbitrary padding(任意填充):人为设定填充。..._cnn “相同填充”(same padding)
文章浏览阅读790次,点赞29次,收藏28次。手绘了下图所示的kafka知识大纲流程图(xmind文件不能上传,导出图片展现),但都可提供源文件给每位爱学习的朋友一个人可以走的很快,但一群人才能走的更远。不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人,都欢迎扫码加入我们的的圈子(技术交流、学习资源、职场吐槽、大厂内推、面试辅导),让我们一起学习成长![外链图片转存中…(img-Qpoc4gOu-1712656009273)][外链图片转存中…(img-bSWbNeGN-1712656009274)]
文章浏览阅读469次。Date对象取得年份有getYear和getFullYear两种方法经 测试var d=new Date;alert(d.getYear())在IE中返回 2009,在Firefox中会返回109。经查询手册,getYear在Firefox下返回的是距1900年1月1日的年份,这是一个过时而不被推荐的方法。而alert(d.getFullYear())在IE和FF中都会返回2009。因此,无论何时都应使用getFullYear来替代getYear方法。例如:2016年用 getFullYea_getyear和getfullyear
文章浏览阅读182次。Unix传奇(上篇) 陈皓 了解过去,我们才能知其然,更知所以然。总结过去,我们才会知道我们明天该如何去规划,该如何去走。在时间的滚轮中,许许多的东西就像流星一样一闪而逝,而有些东西却能经受着时间的考验散发着经久的魅力,让人津津乐道,流传至今。要知道明天怎么去选择,怎么去做,不是盲目地跟从今天各种各样琳琅满目前沿技术,而应该是去 —— 认认真真地了解和回顾历史。 Unix是目前还在存活的操作系_unix传奇pdf
文章浏览阅读308次。哈希算法:将字符串映射为数字形式,十分巧妙,一般运用为进制数,进制据前人经验,一般为131,1331时重复率很低,由于字符串的数字和会很大,所以一般为了方便,一般定义为unsigned long long,爆掉时,即为对 2^64 取模,可以对于任意子序列的值进行映射为数字进而进行判断入门题目链接:AC代码:#include<bits/stdc++.h>using na..._ac算法 哈希
文章浏览阅读952次,点赞13次,收藏27次。由于觉得Qt的编辑界面比较丑,所以想用vs2022的编辑器写Qt加MySQL的项目。_在vs中 如何装qt5sqlmysql模块
文章浏览阅读1k次。选择题题目:下面的哪个调研内容属于经济环境调研?()题目:()的目的就是加强与客户的沟通,它是是网络媒体也是网络营销的最重要特性。题目:4Ps策略中4P是指产品、价格、顾客和促销。题目:网络市场调研是目前最为先进的市场调研手段,没有任何的缺点或不足之处。题目:市场定位的基本参数有题目:市场需求调研可以掌握()等信息。题目:在开展企业网站建设时应做好以下哪几个工作。()题目:对企业网站首页的优化中,一定要注意下面哪几个方面的优化。()题目:()的主要作用是增进顾客关系,提供顾客服务,提升企业_画中画广告之所以能有较高的点击率,主要由于它具有以下特点
文章浏览阅读1k次,点赞2次,收藏5次。以爬取CSDN为例子:第一步:导入请求库第二步:打开请求网址第三步:打印源码import urllib.requestresponse=urllib.request.urlopen("https://www.csdn.net/?spm=1011.2124.3001.5359")print(response.read().decode('utf-8'))结果大概就是这个样子:好的,继续,看看打印的是什么类型的:import urllib.requestresponse=urllib.r_urlopen the read operation timed out
文章浏览阅读304次。修正sina.com/sina.cn邮箱获取不到联系人,并精简修改了其他邮箱代码,以下就是升级版版本的介绍:完整版本,整合了包括读取邮箱通讯录、MSN好友列表的的功能,目前读取邮箱通讯录支持如下邮箱:gmail(Y)、hotmail(Y)、 live(Y)、tom(Y)、yahoo(Y)(有点慢)、 sina(Y)、163(Y)、126(Y)、yeah(Y)、sohu(Y) 读取后可以发送邮件(完..._通讯录 应用读取 邮件 的相关
文章浏览阅读213次。云计算及虚拟化教程学习云计算、虚拟化和计算机网络的基本概念。此视频教程共2.0小时,中英双语字幕,画质清晰无水印,源码附件全课程英文名:Cloud Computing and Virtualization An Introduction百度网盘地址:https://pan.baidu.com/s/1lrak60XOGEqMOI6lXYf6TQ?pwd=ns0j课程介绍:https://www.aihorizon.cn/72云计算:概念、定义、云类型和服务部署模型。虚拟化的概念使用 Type-2 Hyperv_云计算与虚拟化技术 教改