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 点击打开链接
文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下:http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态:[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态
文章浏览阅读1.3w次,点赞45次,收藏99次。我个人用的是anaconda3的一个python集成环境,自带jupyter notebook,但在我打开jupyter notebook界面后,却找不到对应的虚拟环境,原来是jupyter notebook只是通用于下载anaconda时自带的环境,其他环境要想使用必须手动下载一些库:1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境
文章浏览阅读5.2k次,点赞19次,收藏28次。选择scoop纯属意外,也是无奈,因为电脑用户被锁了管理员权限,所有exe安装程序都无法安装,只可以用绿色软件,最后被我发现scoop,省去了到处下载XXX绿色版的烦恼,当然scoop里需要管理员权限的软件也跟我无缘了(譬如everything)。推荐添加dorado这个bucket镜像,里面很多中文软件,但是部分国外的软件下载地址在github,可能无法下载。以上两个是官方bucket的国内镜像,所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn
文章浏览阅读4.5k次,点赞2次,收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了,例如:这里的:sty..._vue el-color-picker
文章浏览阅读640次。基于芯片日益增长的问题,所以内核开发者们引入了新的方法,就是在内核中只保留函数,而数据则不包含,由用户(应用程序员)自己把数据按照规定的格式编写,并放在约定的地方,为了不占用过多的内存,还要求数据以根精简的方式编写。boot启动时,传参给内核,告诉内核设备树文件和kernel的位置,内核启动时根据地址去找到设备树文件,再利用专用的编译器去反编译dtb文件,将dtb还原成数据结构,以供驱动的函数去调用。firmware是三星的一个固件的设备信息,因为找不到固件,所以内核启动不成功。_exynos 4412 刷机
文章浏览阅读2w次,点赞24次,收藏42次。Linux系统配置jdkLinux学习教程,Linux入门教程(超详细)_linux配置jdk
文章浏览阅读3.3k次,点赞5次,收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图:_matlab微米怎么输入
文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件,可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中,对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数,而关闭则使用fclose函数。在C语言中,可以使用fread和fwrite函数进行二进制读写。 Biaoge 于2024-03-09 23:51发布 阅读量:7 ️文章类型:【 C语言程序设计 】在C语言中,用于打开文件的函数是____,用于关闭文件的函数是____。
文章浏览阅读3.4k次,点赞2次,收藏13次。跟随鼠标移动的粒子以grid(SOP)为partical(SOP)的资源模板,调整后连接【Geo组合+point spirit(MAT)】,在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in(CHOP)鼠标位置到metaball的坐标,实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动
文章浏览阅读178次。项目运行环境配置:Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX(Webstorm也行)+ Eclispe(IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持)。项目技术:Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成,B/S模式 + Maven管理等等。环境需要1.运行环境:最好是java jdk 1.8,我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计
文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多,会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析,后面会慢慢更新。另外,博客只作为自己学习记录的一种方式,对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析 时序图
文章浏览阅读2.4k次,点赞41次,收藏13次。java 数据结构与算法 ——快速排序法_快速排序法