Spark系列之Spark启动与基础使用

news2025/1/17 23:20:47

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 \
--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

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

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();
    }
}

3.4.2 Java8 Lambda表达式版本WordCount

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();

    }
}

3.4.3 Scala版本WordCount

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()
  }
}

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

微博地址: http://weibo.com/luoyepiaoxue2014 点击打开链接

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/44589.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

长短期记忆网络(LSTM)重点!(自己汇集了很多资料在这篇博客)

文章目录参考资料推荐基础知识评论区精髓代码实现底层实现简洁实现参考资料推荐 心心念念 学了这么久 &#xff0c;终于学到第57集了。 参考一篇掘金的图文LSTM 李宏毅老师的手撕视频配套课件 27:39 开始手撕 看完了李沐老师的LSTM又去找了李宏毅老师的课程然后发现又多了个导…

西门子机床联网

一、设备信息确认 1、确认型号 数控面板拍照确认&#xff1a; 此系统为&#xff1a;西门子828D 还有一种情况是面板无任何版本信息&#xff0c;这时就需要进入系统里面再确认。 2、确认通讯接口 1、数控面板的后面 X130为网络标号 2、其他位置 其他位置一般是前面位置用…

H3C mstp+vrrp实验 新华三杯拆解

H3C mstpvrrp实验一、实验拓扑二、实验要求局域网规划&#xff1a;可靠性&#xff1a;三、实验配置&#xff08;一&#xff09;链路聚合1.创建链路聚合组2.检查&#xff08;二&#xff09;VLAN1.创建vlan2.放行vlan3.检查&#xff08;三&#xff09;MSTP1.配置MSTP域2.配置主备…

通俗易懂的java设计模式(1)-单例模式

什么是单例模式&#xff1f; 单例模式是java中最简单的一种设计模式 需要注意的问题&#xff1a; 1.单例类有且只能有一个实例 2.单例类必须自己创建出这个实例&#xff0c;并提供给外界 那么如何自己创建实例而不让外界创建呢&#xff1f;很简单&#xff0c;我们将无参的构造函…

麦芽糖-刀豆球蛋白A,maltose-ConcanavalinA,刀豆球蛋白A-PEG-麦芽糖

麦芽糖-刀豆球蛋白A,maltose-ConcanavalinA,刀豆球蛋白A-PEG-麦芽糖 中文名称&#xff1a;麦芽糖-刀豆球蛋白A 英文名称&#xff1a;maltose-ConcanavalinA 别称&#xff1a;刀豆球蛋白A修饰麦芽糖&#xff0c;ConA-麦芽糖 还可以提供PEG接枝修饰麦芽糖&#xff0c;麦芽糖…

SpringCloud-alibaba-Nacos 从理论到落地使用

Nacos: Dynamic Naming and Configuration Service 就是&#xff1a; 一个更易于构建云原生应用的动态服务发现、配置管理和服务管理平台。、 下面是生态图&#xff1a; Nacos EurekaConfig Bus 即 Nacos就是注册中心 配置中心的组合 他能干什么呢&#xff1f; 1、替代Eu…

防火墙firewalld

RHEL7中有几种防火墙共存&#xff1a;firewalld、iptables、ebtables等。基于iptables的防火墙默认不启动&#xff0c;但仍然可以继续使用。RHEL7默认使用firewalld作为防火墙&#xff0c;管理工具是firewall-cmd。RHEL7的内核版本是3.10&#xff0c;在此版本的内核里防火墙的包…

论文阅读2 Learning Hierarchy-Aware Knowledge Graph Embeddings for Link Prediction

目录 问题 创新 1、Introduction 2、相关工作 3、HAKE模型 原文&#xff1a;[1911.09419] Learning Hierarchy-Aware Knowledge Graph Embeddings for Link Prediction (arxiv.org) 问题 现有的知识图嵌入模型主要关注对称/反对称、反转和复合symmetry/antisymmetry, i…

EquiVSet

又搬来一个于最优子集的神经集合函数学习方法 集合函数被广泛应用于各种场景之中&#xff0c;例如商品推荐、异常检测和分子筛选等。在这些场景中&#xff0c;集合函数可以被视为一个评分函数&#xff1a;其将一个集合作为输入并输出该集合的分数。我们希望从给定的集合中选取…

【新知实验室-TRTC开发】实时音视频之欢度世界杯

目录 一、什么是TRTC 二、用5分钟跑通一个demo 1、开通腾讯云-TRTC 2、获取demo必须的两把钥匙 2.1输入应用名称 2.2下载对应的源码包&#xff08;手机、web、小程序等&#xff09; 2.3拿到钥匙 2.4完成 三、搭建一起看世界杯应用 1、解压源码&#xff08;耗时30S&#x…

Linux下top命令详解

Linux下top命令用法详解 作为一名Linux软件攻城狮&#xff0c;top命令大家应该并不陌生。top命令是Linux下常用的性能分析工具&#xff0c;能够实时显示系统中各个进程的资源占用状况。top可以动态显示过程,不断刷新当前状态。top命令提供了实时的对系统处理器的状态监视。它将…

Kotlin高仿微信-第5篇-主页-通讯录

Kotlin高仿微信-项目实践58篇详细讲解了各个功能点&#xff0c;包括&#xff1a;注册、登录、主页、单聊(文本、表情、语音、图片、小视频、视频通话、语音通话、红包、转账)、群聊、个人信息、朋友圈、支付服务、扫一扫、搜索好友、添加好友、开通VIP等众多功能。 Kotlin高仿…

【博客547】keepalived实现vip的原理剖析

keepalived实现vip的原理剖析 keepalived实现vip的原理&#xff1a;vrrp gratuitous arp 1、vrrp vrrp更多细节参考&#xff1a;vrrp技术白皮书 相关术语&#xff1a; VRRP工作过程 VRRP的工作过程为&#xff1a; 1、虚拟路由器中的路由器根据优先级选举出Master。Master…

【Python】五、程序循环结构

文章目录实验目的一、掌握while语句二、掌握for循环和range()内建函数三、掌握循环语句嵌套四、掌握break语句和continue语句五、编写程序实现猜数字的游戏1.设计思路2.设计算法3.参考代码4.实验截图实验目的 掌握循环结构&#xff1b;培养学生动手查阅资料能力和解决实际问题的…

matlab实现线性参数的最小二乘法处理

一、实验目的 最小二乘法原理是一种在多学科领域中获得广泛应用的数据处理方法。通过实验要求掌握最小二乘法基本原理、正规方程以及组合测量的最小二乘法处理办法。 二、实验原理 &#xff08;1&#xff09;测量结果的最可信赖值应在残余误差平方和为最小的条件下求出&#…

破局模块总结 -- 宁向东的清华管理学课总结

1. 管理学就是要破局而出 为什么需要管理学&#xff1f;德鲁克说&#xff0c;我们需要选择正确的事情去做&#xff0c;并且把事情作对。 管理学是一个分析、权衡和决策的学问&#xff0c;分析、权衡和选择的目的是什么&#xff1f;就是要破局而出。 管理學是研究人類管理活動…

《Linux运维总结:基于快照模式迁移单节点elasticsearch数据(方案二)》

一、背景信息 说明&#xff1a;由于整个系统需要从互联网迁移到政务外网&#xff0c;elasticsearch作为其中一个组件&#xff0c;也需要将 所有索引数据 迁移到政务外网。 由于数据量比较大&#xff0c;所以使用快照的模式对elasticsearch数据进行备份及恢复操作&#xff0c;提…

[附源码]计算机毕业设计springboot病人跟踪治疗信息管理系统

项目运行 环境配置&#xff1a; Jdk1.8 Tomcat7.0 Mysql HBuilderX&#xff08;Webstorm也行&#xff09; Eclispe&#xff08;IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持&#xff09;。 项目技术&#xff1a; SSM mybatis Maven Vue 等等组成&#xff0c;B/S模式 M…

接口测试当中的权限限制测试和状态机测试【杭州多测师_王sir】【杭州多测师】...

一、权限限制 权限限制我需要单独拎出来讲&#xff0c;它很重要&#xff01;很多系统中都存在系统管理员、普通用户等不同角色的用户&#xff0c;系统管理员拥有一些普通用户没有的权限&#xff0c;比如系统管理员可以发布公告&#xff0c;而普通用户则只能查看&#xff0c;系统…

【MATLAB教程案例48】初识点云——pcshow,pointCloud,pcwrite,pcread,pcdenoise等点云基本操作函数学习

欢迎订阅《FPGA学习入门100例教程》、《MATLAB学习入门100例教程》 本课程效果预览 目录 1.软件版本 2.什么是点云?