大数据开发之Spark（入门）

第 1 章：Spark概述

1.1 什么是spark

回顾：hadoop主要解决，海量数据的存储和海量数据的分析计算。
spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。

1.2 hadoop与spark历史

hadoop的yarn框架比spark框架诞生的晚，所以spark自己也涉及了一套资源调度框架。
区别：
1、mr是基于磁盘的，spark是基于内存
2、mr的task是进程
3、spark的task是线程，在executor进程里执行的是线程
4、mr在container里执行（留有接口方便插入），spark在worker里执行（自己用，没有接口）
5、mr适合做一次计算，spark适合做迭代计算

1.3 hadoop与spark框架对比

1、hadoop mr框架
从数据源获取数据，经过分析计算，将结果输出到指定位置，核心是一次计算，不适合迭代计算。
在这里插入图片描述

2、spark框架
spark框架计算比mr快的原因是：中间结果不落盘。注意spark的shuffle也是落盘的。
在这里插入图片描述

1.4 spark内置模块

在这里插入图片描述

spark core：实现了spark的基本功能，包括任务调度、内存管理、错误恢复、与存储系统交互等模块。spark core中还包含了对弹性分布式数据集（resilient distributed dataset，简称rdd）的api定义。
spark sql：是spark用来操作结构化数据的程序包。通过spark sql，我们可以使用sql或者apache hive版本的hql来查询数据。spark sql支持多种数据源，比如hive表、parquet以及json等。
spark mllib：提供常见的机器学习功能的程序库。包括分类、回归、聚类、协同过滤等，还提供了模型评估、数据导入等额外的支持功能。
spark graphx：主要用于图形并行计算和图挖掘系统的组件。
集群管理器：spark设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。为了实现这样的要求，同时获得最大灵活性，spark支持在各种集群管理器（cluster manager）上运行，包括hadoop yarn、apache mesos，以及spark自带的一个简易调度器，叫做独立调度器。

1.5 spark特点

1、快：与hadoop的mapreduce相比，spark基于内存的运算要快上100倍以上，基于硬盘的运算也要快10倍以上。spark实现了高效的dag执行引擎，口头语通过基于内存来高效处理数据流。计算的中间结果是存在于内存中的。
2、易用：spark支持java、python和scala的api，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且spark支持交互式的python和scala的shell，可以非常方便地在这些shell种使用spark集群来验证解决问题的方法。
3、通用：spark提供了统一的解决方案。spark可以用于，交互式查询（spark sql）、实时流处理（spark streaming）、机器学习（spark mllib）和图计算（graphx）。这些不同类型的处理1都可以在同一个应用种无缝使用。减少了开发和维护的人力成本和部署平台的物力成本。
4、兼容性：spark可以非常方便地与其它地开源产品进行融合。比如：spark可以使用hadoop的yarn和apache mesos作为它的资源管理和调度器，并且可以处理所有hadoop支持的数据，包括hdfs、hbase等。这对于已经部署hadoop集群的用户特别重要，因为不需要做任何数据迁移就可以使用spark的强大处理能力。

第 2 章：spark运行模式

部署spark集群大体上分为两种模式：单机模式与集群模式
大多数分布式框架都支持单机模式，方便开发者调试框架的运行环境。但是在生产环境种，并不会使用单机模式。因此，后续直接按照集群模式部署spark集群。
下面详细列举了spark目前支持的部署模式。
1、local模式：在本地部署spark服务
2、standalone模式：spark自带的任务调度模式。（国内常用）
3、yarn模式：spark使用hadoop的yarn组件进行资源和任务调度。（国内最常用）
4、mesos模式：spark使用mesos平台进行资源与任务的调度。（国内很少用）

2.2 local模式

local模式就是运行在一台计算机上的模式，通常就是用于在本机上练手和测试

2.2.1 安装使用

1）上传并解压spark安装包

[atguigu@hadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-local

2）官方求pi案例

[atguigu@hadoop102 spark-local]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[2] \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

可以查看spark-submit所用参数

[atguigu@hadoop102 spark-local]$ bin/spark-submit

–class：表示要执行程序的主类
–master local[2]“
（1）local：没有指定线程数，则所有计算都运行在一个线程当中，没有任何并行计算。
（2）local[k]：指定使用k个core来运行计算，比如local[2]就是运行2个core来执行

20/09/20 09:30:53 INFO TaskSetManager:
20/09/15 10:15:00 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
20/09/15 10:15:00 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)

（3）local[*]：默认模式。自动帮你按照cpu最多核来设置线程数。比如cpu有8核，spark帮你自动设置8个线程。

20/09/20 09:30:53 INFO TaskSetManager:
20/09/15 10:15:58 INFO Executor: Running task 1.0 in stage 0.0 (TID 1)
20/09/15 10:15:58 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
20/09/15 10:15:58 INFO Executor: Running task 2.0 in stage 0.0 (TID 2)
20/09/15 10:15:58 INFO Executor: Running task 4.0 in stage 0.0 (TID 4)
20/09/15 10:15:58 INFO Executor: Running task 3.0 in stage 0.0 (TID 3)
20/09/15 10:15:58 INFO Executor: Running task 5.0 in stage 0.0 (TID 5)
20/09/15 10:15:59 INFO Executor: Running task 7.0 in stage 0.0 (TID 7)
20/09/15 10:15:59 INFO Executor: Running task 6.0 in stage 0.0 (TID 6)

3）结果展示
该算法是利用蒙特-卡罗算法求pi
在这里插入图片描述

2.2.2 官方wordcount案例

1、需求：读取多个输入文件，统计每个单词出现的总次数。
2、需求分析
在这里插入图片描述

3、代码实现
1）准备文件

[atguigu@hadoop102 spark-local]$ mkdir input

在Input下创建2个文件1.txt和2.txt，并输入一下内容

hello atguigu
hello spark

2）启动spark-shell

[atguigu@hadoop102 spark-local]$ bin/spark-shell20/07/02 10:17:11 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
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://hadoop102:4040
Spark context available as 'sc' (master = local[*], app id = local-1593656236294).
Spark session available as 'spark'.
Welcome to____              __/ __/__  ___ _____/ /___\ \/ _ \/ _ `/ __/  '_//___/ .__/\_,_/_/ /_/\_\   version 3.1.3/_/Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_212)
Type in expressions to have them evaluated.
Type :help for more information.scala>

注意：sc是sparkcore程序的入口；spark是sparksql程序入口；master=local[*]表示本地模式运行。
3）再开启一个hadoop102远程连接窗口，发现了一个sparksubmit进程

[atguigu@hadoop102 spark-local]$ jps
3627 SparkSubmit
4047 Jps

运行任务方式说明：spark-submit，是将jar上传到集群，执行spark任务；spark-shell，相当于命令行工具，本身也是一个application。
4）登录hadoop102:4040，查看程序运行情况
在这里插入图片描述

说明：本地模式下，默认的调度器为fifo。
5）运行workcount程序

scala>sc.textFile("/opt/module/spark-local/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collectres0: Array[(String, Int)] = Array((hello,4), (atguigu,2), (spark,2))

注意：只有collect开始执行时，才会加载数据
可登录hadoop102:4040查看程序运行结果
在这里插入图片描述

2.3 standalone模式

standalone模式是spark自带的资源调度引擎，构建一个由master+worker构成的spark集群，spark运行在集群种。
这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群，不需要借助hadoop的yarn和mesos等其它框架。

2.3.1 master和worker集群资源管理

master：spark特有资源调度系统的leader。掌管着整个集群的资源信息，类似于yarn框架中的resourcemanager。
worker：spark特有资源调度系统的slave，有多个。每个slave掌管着所在节点的资源信息，类似于yarn框架中的nodemanager。
在这里插入图片描述

master和worker是spark的守护进程、集群资源管理者，即spark在特定模式（standalone）下正常运行必须要有的后台常驻进程。

2.3.2 driber和executor任务的管理者

在这里插入图片描述

driver和executor是临时程序，当有具体任务提交到spark集群才会开启的程序。standalone模式是spark自带的资源调度引擎，构建一个由master+worker构成spark集群，spark运行在集群中。
这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群，不需要借助hadoop的yarn和mesos等其它框架。

2.3.2 安装使用

1、集群规划
在这里插入图片描述

2、再解压一份spark安装包，并修改解压后的文件夹名称为spark-standalone

[atguigu@hadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-standalone

3、进入spark的配置文件/opt/module/spark-standalone/conf

[atguigu@hadoop102 spark-standalone]$ cd conf

4、修改slave文件，添加work节点

atguigu@hadoop102 conf]$ mv slaves.template slaves
[atguigu@hadoop102 conf]$ vim slaves
hadoop102
hadoop103
hadoop104

5、修改spark-env.sh文件，添加master节点

[atguigu@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguigu@hadoop102 conf]$ vim spark-env.shSPARK_MASTER_HOST=hadoop102
SPARK_MASTER_PORT=7077

6、分发spark-standalone包

[atguigu@hadoop102 module]$ xsync spark-standalone/

7、启动spark集群

[atguigu@hadoop102 module]$ xsync spark-standalone/

查看三台服务器运行进程（xcall.sh是以前数仓项目里面讲的脚本）

[atguigu@hadoop102 spark-standalone]$ xcall.sh jps
================atguigu@hadoop102================
3238 Worker
3163 Master
================atguigu@hadoop103================
2908 Worker
================atguigu@hadoop104================
2978 Worker

注意：如果遇见”Java_home not set“异常，可以在sbin目录下的spark-config.sh文件中加入如下配置

export JAVA_HOME=XXXX

8、网页查看：hadoop102:8080
9、官方求pi案例

[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

参数：–master spark://hadoop102:7077指定要连接的集群的master。
10、页面查看http://hadoop102:8080/，发现执行本次任务，默认采用三台服务器节点的总核数24核，每个节点内存1024M.
8080：master的webui
4040：application的webui的端口号
在这里插入图片描述

2.3.3 参数说明

1、配置executor可用内存为2G，使用cpu核数为2个

[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 2G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

2、页面查看http://hadoop102:8080/
在这里插入图片描述

3、基本语法

bin/spark-submit \
--class <main-class>
--master <master-url> \
... # other options
<application-jar> \
[application-arguments]

4、参数说明
在这里插入图片描述

2.3.4 配置历史服务

由于spark-shell停止掉后，hadoop102:4040页面就看不到历史任务的运行情况，所以开发时都配置历史服务器记录任务运行情况
1、修改spark-default.conf.template名称

[atguigu@hadoop102 conf]$ mv spark-defaults.conf.template spark-defaults.conf

2、修改spark-default.conf文件，配置日志存储路径

[atguigu@hadoop102 conf]$ vim spark-defaults.conf
spark.eventLog.enabled          true
spark.eventLog.dir              hdfs://hadoop102:8020/directory

注意：需要启动hdaoop集群，hdfs上的目录需要提前存在

[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop102 hadoop-3.1.3]$ hadoop fs -mkdir /directory

3、修改spark-env.sh文件，添加如下配置

[atguigu@hadoop102 conf]$ vim spark-env.shexport SPARK_HISTORY_OPTS="
-Dspark.history.ui.port=18080 
-Dspark.history.fs.logDirectory=hdfs://hadoop102:8020/directory 
-Dspark.history.retainedApplications=30"

1）参数1含义：webui访问的端口号为18080
2）参数2含义：指定历史服务器日志存储路径（读）
3）参数3含义：指定保存application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上的显示的应用数
4、分发配置文件

[atguigu@hadoop102 conf]$ xsync spark-defaults.conf spark-env.sh

5、启动历史服务

[atguigu@hadoop102 spark-standalone]$ 
sbin/start-history-server.sh

6、再次执行任务

[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077 \
--executor-memory 1G \
--total-executor-cores 2 \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

7、查看spark历史服务地址：hadoop102:18080
在这里插入图片描述

2.3.5 配置高可用（HA）

1、高可用原理
在这里插入图片描述

2、配置高可用
1）停止集群

[atguigu@hadoop102 spark-standalone]$ sbin/stop-all.sh

2）zookeeper正常安装并启动（基于以前讲的数仓项目脚本）

[atguigu@hadoop102 zookeeper-3.4.10]$ zk.sh start

3）修改spark-env.sh文件添加如下配置

[atguigu@hadoop102 conf]$ vim spark-env.sh#注释掉如下内容：
#SPARK_MASTER_HOST=hadoop102
#SPARK_MASTER_PORT=7077#添加上如下内容。配置由Zookeeper管理Master，在Zookeeper节点中自动创建/spark目录，用于管理：
export SPARK_DAEMON_JAVA_OPTS="
-Dspark.deploy.recoveryMode=ZOOKEEPER 
-Dspark.deploy.zookeeper.url=hadoop102,hadoop103,hadoop104 
-Dspark.deploy.zookeeper.dir=/spark"#添加如下代码
#Zookeeper3.5的AdminServer默认端口是8080，和Spark的WebUI冲突
export SPARK_MASTER_WEBUI_PORT=8989

4）分发配置文件

[atguigu@hadoop102 conf]$ xsync spark-env.sh

5）在hadoop102上启动全部节点

[atguigu@hadoop102 spark-standalone]$ sbin/start-all.sh

6）在hadoop103上单独启动master节点

[atguigu@hadoop103 spark-standalone]$ sbin/start-master.sh

7）在启动一个hadoop102窗口，将/opt/module/spark-local/input数据上传到hadoop集群的/input目录

[atguigu@hadoop102 spark-standalone]$ hadoop fs -put /opt/module/spark-local/input/ /input

8）spark ha集群访问

[atguigu@hadoop102 spark-standalone]$
bin/spark-shell \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2g \
--total-executor-cores 2

参数：–master spark://hadoop102:7077指定要连接的集群的master
注：一旦配置了高可用以后，master后面要连接多个master
9）执行wordcount程序

scala>sc.textFile("hdfs://hadoop102:8020/input").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collectres0: Array[(String, Int)] = Array((hello,4), (atguigu,2), (spark,2))

3、高可用性测试
在这里插入图片描述

1）查看hadoop102的master进程

[atguigu@hadoop102 ~]$ jps
5506 Worker
5394 Master
5731 SparkSubmit
4869 QuorumPeerMain
5991 Jps
5831 CoarseGrainedExecutorBackend

2）kill掉hadoop102的master进程，页面中观察http://hadoop103:8080/的状态是否切换为active

[atguigu@hadoop102 ~]$ kill -9 5394

3）再启动hadoop102的master进程

[atguigu@hadoop102 spark-standalone]$ sbin/start-master.sh

2.3.5 运行流程

spark由standalone-client核standalone-cluster两种模式，主要区别在于：driver程序的运行节点。
1、客户端模式

[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2G \
--total-executor-cores 2 \
--deploy-mode client \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

–deploy-mode client，表示driver程序运行再本地客户端，默认模式。
standalone client运行流程
在这里插入图片描述

2、集群模式

[atguigu@hadoop102 spark-standalone]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop102:7077,hadoop103:7077 \
--executor-memory 2G \
--total-executor-cores 2 \
--deploy-mode cluster \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

–deploy-mode cluster，表示driver程序运行在集群
standalone cluster运行流程
在这里插入图片描述

1）查看http://hadoop102:8989/页面，点击completed drivers里面的worker
在这里插入图片描述

2）跳转到spark worker页面，点击finished drivers中logs下面的stdout
在这里插入图片描述

3）最终打印结果如下
在这里插入图片描述

注意：在测试standalone模式，cluster运行流程的时候，阿里云用户访问不到worker，因为worker是从master内部跳转的，这是正常的，实际工作中我们不可能通过客户端访问的，这些恶端口都对外都会禁用，需要的时候会通过授权到master访问worker

2.4 yarn模式（重点）

spark客户端直接连接yarn，不需要额外构建spark集群

2.4.1 安装使用

1、停止standalone模式下的spark集群

[atguigu@hadoop102 spark-standalone]$ sbin/stop-all.sh
[atguigu@hadoop102 spark-standalone]$ zk.sh stop
[atguigu@hadoop103 spark-standalone]$ sbin/stop-master.sh

2、为了防止和standalone模式冲突，再单独解压一份spark

[atguigu@hadoop102 software]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/

3、进入到/opt/module目录，修改spark-~名称为spark-yarn

[atguigu@hadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2/ spark-yarn

4、修改hadoop配置文件/opt/module/~/yarn-site.xml，添加如下内容
因为测试环境虚拟机内存较少，防止执行过程进行倍意外杀死，做如下处理

[atguigu@hadoop102 hadoop]$ vim yarn-site.xml
<!--是否启动一个线程检查每个任务正使用的物理内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property><name>yarn.nodemanager.pmem-check-enabled</name><value>false</value>
</property><!--是否启动一个线程检查每个任务正使用的虚拟内存量，如果任务超出分配值，则直接将其杀掉，默认是true -->
<property><name>yarn.nodemanager.vmem-check-enabled</name><value>false</value>
</property>

5、分发配置文件

[atguigu@hadoop102 conf]$ xsync /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml

6、修改/opt/~/spark-env.sh，添加yarn_conf_dir配置，保证后续运行任务的路径都编程集群路径

[atguigu@hadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguigu@hadoop102 conf]$ vim spark-env.shYARN_CONF_DIR=/opt/module/hadoop-3.1.3/etc/hadoop

7、启动hdfs以及yarn集群

[atguigu@hadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguigu@hadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh

8、执行一个程序

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

参数：–master yarn，表示yarn方式运行；–deploy-mode，表示客户端方式运行程序
9、查看hadoop103:8088页面，点击history，查看历史页面
在这里插入图片描述

2.4.2 配置历史服务

由于是重新解压的spark压缩文件，所以需要针对yarn模式，再次配置一下历史服务器。
1、修改spark-default.conf.template名称

2、修改spark-default.conf文件，配置日志存储路径（写）

3、修改spark-env.sh文件，添加如下配置

参数1含义：webui访问的端口号为18080
参数2含义：指定历史服务器日志存储路径（读）
参数3含义：指定保存application历史记录的个数，如果超过这个值，旧的应用程序信息将被删除，这个是内存中的应用数，而不是页面上显示的应用数

2.4.3 配置查看历史日志

为了能从yarn上关联到spark历史服务器，需要配置spark历史服务器关联路径
目的：点击yarn（8088）上spark任务的history按钮，进入的是spark历史服务器（18080），而不再是yarn历史服务器（19888）
1、修改配置文件/opt/module/~/spark-defaults.conf
添加如下内容：

spark.yarn.historyserver.address=hadoop102:18080
spark.history.ui.port=18080

2、重启spark历史服务

[atguigu@hadoop102 spark-yarn]$ sbin/stop-history-server.sh [atguigu@hadoop102 spark-yarn]$ sbin/start-history-server.sh

3、提交任务到yarn执行

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

4、web页面查看日志：http://hadoop103:8088/cluster
在这里插入图片描述

点击”history“跳转到http://hadoop102:18080/
在这里插入图片描述

2.4.4 运行流程

spark由yarn-client和yarn-cluster两种模式，主要区别在于：driver程序的运行节点
yarn-client：driver程序运行在客户端，适用于交互、调试，希望立即看到app的输出
yarn-cluster：driver程序运行在由resourcemanager启动的appmaster，适用于生产环境
1、客户端模式（默认）

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

yarnclient运行模式介绍
在这里插入图片描述

2、集群模式

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode cluster \
./examples/jars/spark-examples_2.12-3.1.3.jar \
10

（1）查看http://hadoop103:8088/cluster页面，点击history按钮，跳转到历史详情页面
在这里插入图片描述

（2）http://hadoop102:18080点击executors->点击driver的stdout
在这里插入图片描述

注意：如果在yarn日志端无法查看到具体的日志，则在yarn-site.xml中添加如下配置并启动yarn历史服务器
在这里插入图片描述

<property><name>yarn.log.server.url</name><value>http://hadoop102:19888/jobhistory/logs</value>
</property>

注意：hadoop历史服务器也要启动 mr-jobhistory-daemon.sh start historyserver
yarncluster模式
在这里插入图片描述

2.6 几种模式对比

在这里插入图片描述

2.7 端口号总结

1、spark查看当前spark-shell运行任务情况端口号：4040
2、spark master内部通信服务端口号：7077（类似于yarn的8032（rm和nm的内部通信）端口）
3、spark standalone模式master web端口号：8080（类似于hadoop yarn任务运行情况查看端口号：8088）（yarn模式）8989
4、spark历史服务器端口号：18080（类似于hadoop历史服务器端口号：19888）

第 3 章：workcount案例实操

spark shell仅在测试和验证我们的程序时使用的较多，在生产环境中，通常会在idea中编制程序，然后打包jar包，然后提交到集群，最常用的是创建一个maven项目，利用maven来管理jar包的依赖。

3.1 部署环境

1、创建一个maven项目wordcount
2、在项目wordcount上点击右键，add framework support -> 勾选scala
3、在main下创建scala文件夹，并右键mark directory as sources root -> 在scala下创建包com.atguigu.spark
4、输入文件夹准备
在这里插入图片描述
在这里插入图片描述

5、导入项目依赖
下方的的是scala语言打包插件，只要使用scala语法打包运行到linux上面，必须要有

<dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.1.3</version></dependency>
</dependencies><build><finalName>WordCount</finalName><plugins><plugin><groupId>net.alchim31.maven</groupId><artifactId>scala-maven-plugin</artifactId><version>3.4.6</version><executions><execution><goals><goal>compile</goal><goal>testCompile</goal></goals></execution></executions></plugin></plugins>
</build>

3.2 本地调试

本地spark程序调试需要使用local提交模式，即将本机当作运行环境，master和worker都为本机。运行时直接加断点调试即可。如下：
1、代码实现

package com.atguigu.sparkimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit = {//1.创建SparkConf并设置App名称val conf = new SparkConf().setAppName("WC").setMaster("local[*]")//2.创建SparkContext，该对象是提交Spark App的入口val sc = new SparkContext(conf)//3.读取指定位置文件:hello atguigu atguiguval lineRdd: RDD[String] = sc.textFile("input")//4.读取的一行一行的数据分解成一个一个的单词（扁平化）(hello)(atguigu)(atguigu)val wordRdd: RDD[String] = lineRdd.flatMap(_.split(" "))//5. 将数据转换结构：(hello,1)(atguigu,1)(atguigu,1)val wordToOneRdd: RDD[(String, Int)] = wordRdd.map((_, 1))//6.将转换结构后的数据进行聚合处理 atguigu:1、1 =》1+1  (atguigu,2)val wordToSumRdd: RDD[(String, Int)] = wordToOneRdd.reduceByKey(_+_)//7.将统计结果采集到控制台打印wordToSumRdd.collect().foreach(println)//8.关闭连接sc.stop()}
}

2、调试流程
在这里插入图片描述

spark程序运行过程中会打印大量的执行日志，为了能够更好的查看程序的执行结果，可以在项目的resources目录中创建log4j.properties文件，并添加日志配置文件：

log4j.rootCategory=ERROR, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.target=System.err
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n# Set the default spark-shell log level to ERROR. When running the spark-shell, the
# log level for this class is used to overwrite the root logger's log level, so that
# the user can have different defaults for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.Main=ERROR# Settings to quiet third party logs that are too verbose
log4j.logger.org.spark_project.jetty=ERROR
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycle=ERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyper=ERROR
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreter=ERROR
log4j.logger.org.apache.parquet=ERROR
log4j.logger.parquet=ERROR# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandler=FATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistry=ERROR

3、集群运行

3.3 集群运行

1、修改代码，修改运行模式，将输出的方法修改为落盘，同时设置可以自定义的传入传出路径

package com.atguigu.sparkimport org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object WordCount {def main(args: Array[String]): Unit = {// 创建配置对象 添加配置参数val conf: SparkConf = new SparkConf().setAppName("wc")// 如果是yarn模式  写yarn// 如果是本地模式一定要写local.setMaster("yarn")// 初始化scval sc = new SparkContext(conf)// 编写wordCount计算流程// 把读入和写出的路径  做成动态的参数  可以由用户手动填写// 写成main方法参数val lineRDD: RDD[String] = sc.textFile(args(0))// 切分val wordRDD: RDD[String] = lineRDD.flatMap(_.split(" "))// 转换val tupleOneRDD: RDD[(String, Int)] = wordRDD.map((_, 1))// 聚合val wordCountRDD: RDD[(String, Int)] = tupleOneRDD.reduceByKey(_ + _)// 触发计算  一定要使用行动算子// 将结果保存到文件中// 不能重复写入同一个路径wordCountRDD.saveAsTextFile(args(1))}
}

2、打包到集群测试
1）点击package打包，然后，查看打包完后的jar包
在这里插入图片描述

2）将wordcount.jar上传到/opt/module/spark-yarn目录
3）在hdfs上创建，存储输入文件的路径/input

[atguigu@hadoop102 spark-yarn]$ hadoop fs -mkdir /input

4）上传输入文件到/input路径

[atguigu@hadoop102 spark-yarn]$ hadoop fs -put /opt/module/spark-local/input/1.txt /input

5）执行任务

[atguigu@hadoop102 spark-yarn]$ bin/spark-submit \
--class com.atguigu.spark.WordCount \
--master yarn \
./WordCount.jar \
hdfs://hadoop102:8020/input \
hdfs://hadoop102:8020/output

注意：input和output都是hdfs上的集群路径
6）查看运行结果

[atguigu@hadoop102 spark-yarn]$ hadoop fs -cat /output/*

3.4 关联源码

1、按住ctrl键，点击rdd
在这里插入图片描述

2、提示下载或者绑定源码
在这里插入图片描述

3、解压资料包中spark-3.1.3.tgz到非中文路径。例如解压到：e:\02_software
4、点击attach source…按钮，选择源码路径e:\02_software\spark-3.1.3

3.5 异常处理

如果本机操作系统是windows，如果在程序中使用了hadoop相关的东西，比如写入文件到hdfs，则会遇到如下异常：
在这里插入图片描述

出现这个问题的原因，并不是程序的错误，而是用到了hadoop相关的服务，解决办法
1、配置hadoop_home环境变量
2、在idea中配置 run configuration，添加hadoop_home变量
在这里插入图片描述

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

CSS实现文本和图片无限滚动动画

CSS实现文本和图片无限滚动动画

Demo图如下&#xff1a; <style>* {margin: 0;padding: 0;box-sizing: border-box;font-family: Poppins, sans-serif;}body {min-height: 100vh;background-color: rgb(11, 11, 11);color: #fff;display: flex;flex-direction: column;justify-content: center;align-i…
阅读更多...
Vision Mamba:将Mamba应用于计算机视觉任务的新模型

Vision Mamba:将Mamba应用于计算机视觉任务的新模型

Mamba是LLM的一种新架构&#xff0c;与Transformers等传统模型相比&#xff0c;它能够更有效地处理长序列。就像VIT一样现在已经有人将他应用到了计算机视觉领域&#xff0c;让我们来看看最近的这篇论文“Vision Mamba: Efficient Visual Representation Learning with Bidirec…
阅读更多...
Kubernetes operator（一）client-go篇【更新中】

Kubernetes operator（一）client-go篇【更新中】

云原生学习路线导航页&#xff08;持续更新中&#xff09; 本文是 Kubernetes operator学习 系列第一篇&#xff0c;主要对client-go进行学习&#xff0c;从源码阅读角度&#xff0c;学习client-go各个组件的实现原理、如何协同工作等参考视频&#xff1a;Bilibili 2022年最新k…
阅读更多...
pod的亲和性和反亲和性

pod的亲和性和反亲和性

pod的亲和性和反亲和性 调度策略&#xff1a; 匹配标签 操作符 拓扑域 调度目标 node的亲和性 主机标签 In Notin exists doesexists Gt Lt 不支持 指定主机 pod的亲和性 pod的标签 In Notin exists doesexists 支持 pod和指…
阅读更多...
什么是 Web3.0

什么是 Web3.0

什么是Web3.0 对于 Web3.0 的解释网上有很多&#xff0c;目前来说 Web3.0 是一个趋势&#xff0c;尚未有明确的定义。我们今天讨论下几个核心的点&#xff0c;就能很好的理解 Web3.0 要解决哪些问题 谁创造数据&#xff0c;这里的数据可以是一篇博客&#xff0c;一段视频&…
阅读更多...
Oracle Linux 8.9 安装图解

Oracle Linux 8.9 安装图解

风险告知 本人及本篇博文不为任何人及任何行为的任何风险承担责任&#xff0c;图解仅供参考&#xff0c;请悉知&#xff01;本次安装图解是在一个全新的演示环境下进行的&#xff0c;演示环境中没有任何有价值的数据&#xff0c;但这并不代表摆在你面前的环境也是如此。生产环境…
阅读更多...
在Vue.js中，什么是mixins？它们的作用是什么？

在Vue.js中，什么是mixins？它们的作用是什么？

目录 一、Vue.js介绍 二、什么是mixins 三、mixins的应用场景 四、mixins的优势和作用 一、Vue.js介绍 Vue.js是一种流行的JavaScript前端框架,用于构建交互式的Web界面。它被设计为易于理解和集成的框架,使开发者能够快速构建可复用的组件化应用程序。Vue.js采用了MVVM&a…
阅读更多...
如何测试python 版本与 torch 、 torchvision 版本是否对应？

如何测试python 版本与 torch 、 torchvision 版本是否对应？

python 版本与 torch 、 torchvision 版本的对应关系如下图所示&#xff1a; 打开 anaconda powershell prompt&#xff0c;输入如下命令&#xff1a; >python>>>import torch>>>c torch.ones((3,1)) //创建矩阵>>>c c.cuda(0) …
阅读更多...
Keepalived + Nginx双主架构

Keepalived + Nginx双主架构

Keepalived Nginx双主架构 环境准备&#xff1a; keepalived_master1服务器nginx&#xff1a;172.20.26.167 keepalived_master2服务器nginx&#xff1a;172.20.26.198 各服务器关闭selinux、防火墙等服务。 开机安装部署nginx 在172.20.26.167服务器上 yum install ngi…
阅读更多...
分布式深度学习中的数据并行和模型并行

分布式深度学习中的数据并行和模型并行

&#x1f380;个人主页&#xff1a; https://zhangxiaoshu.blog.csdn.net &#x1f4e2;欢迎大家&#xff1a;关注&#x1f50d;点赞&#x1f44d;评论&#x1f4dd;收藏⭐️&#xff0c;如有错误敬请指正! &#x1f495;未来很长&#xff0c;值得我们全力奔赴更美好的生活&…
阅读更多...
LLMs之Vanna：Vanna(利用自然语言查询数据库的SQL工具+底层基于RAG)的简介、安装、使用方法之详细攻略

LLMs之Vanna：Vanna(利用自然语言查询数据库的SQL工具+底层基于RAG)的简介、安装、使用方法之详细攻略

LLMs之Vanna&#xff1a;Vanna(利用自然语言查询数据库的SQL工具底层基于RAG)的简介、安装、使用方法之详细攻略 目录 Vanna的简介 1、用户界面 2、RAG vs. Fine-Tuning 3、为什么选择Vanna&#xff1f; 4、扩展Vanna Vanna的安装和使用方法 1、安装 2、训练 (1)、使用…
阅读更多...
c#中使用UTF-8编码处理多语言文本的有效策略

c#中使用UTF-8编码处理多语言文本的有效策略

使用UTF-8编码处理多语言文本的有效策略 在当今的全球化时代&#xff0c;软件开发者常常需要处理包含多种语言的文本。这不仅涉及英文和其他西方语言&#xff0c;还包括中文、日文、韩文等多字节字符系统。在这篇博客中&#xff0c;我将探讨如何有效地使用UTF-8编码来处理混合语…
阅读更多...
项目管理认证 | 什么是PMP项目管理？PMP证书有什么用？

项目管理认证 | 什么是PMP项目管理？PMP证书有什么用？

01 什么是项目管理&#xff1f; 项目管理&#xff1f;听起来似乎离我们很遥远。其实不然&#xff0c; 学习了项目管理知识后&#xff0c;你会发现&#xff0c;“一切都是项目&#xff0c;一切也将成为项目”。 你可以把港珠澳大桥的建设、开发一款新型手机、开发一个好用的C…
阅读更多...
HarmonyOS 发送http网络请求

HarmonyOS 发送http网络请求

好 本文 我们来说 http请求 首先 我们要操作网络内容 需要申请权限 项目中找到 main目录下的module.json5 最下面加上 "requestPermissions": [{"name": "ohos.permission.INTERNET"} ]这里 我在本地写了一个get接口 大家可以想办法 弄一个后…
阅读更多...
RabbitMQ交换机

RabbitMQ交换机

目录 交换机类型 直连交换机&#xff1a;Direct exchange 主题交换机&#xff1a;Topic exchange 扇形交换机&#xff1a;Fanout exchange 首部交换机&#xff1a;Headers exchange 死信交换机&#xff1a;Dead Letter Exchange 交换机的属性 代码实战 直连&#…
阅读更多...
x-cmd pkg | frp - 内网穿透工具

x-cmd pkg | frp - 内网穿透工具

简介 frp&#xff08;Fast Reverse Proxy&#xff09;是一个专注于内网穿透的高性能反向代理应用&#xff0c;可以将内网服务以安全、便捷的方式通过具有公网 IP 节点的中转暴露到公网。 它采用 C/S 模式&#xff0c;将服务端部署在具有公网 IP 的机器上&#xff0c;客户端部…
阅读更多...
使用torch实现RNN

使用torch实现RNN

在实验室的项目遇到了困难&#xff0c;弄不明白LSTM的原理。到网上搜索&#xff0c;发现LSTM是RNN的变种&#xff0c;那就从RNN开始学吧。 带隐藏状态的RNN可以用下面两个公式来表示&#xff1a; 可以看出&#xff0c;一个RNN的参数有W_xh&#xff0c;W_hh&#xff0c;b_h&am…
阅读更多...
[AutoSar]BSW_OS 06 Autosar OS_Alarms

[AutoSar]BSW_OS 06 Autosar OS_Alarms

一、 目录 一、关键词平台说明一、Timer1.1 配置1.2Periodical Interrupt Timer (PIT)和High Resolution Timer (HRT) 二、Alarm 工作机制三、Code3.1创建一个15ms的runnable3.2mapping到basic task3.3生成代码 关键词 嵌入式、C语言、autosar、OS、BSW 平台说明 项目ValueO…
阅读更多...
k8s的helm

k8s的helm

1、在没有helm之前&#xff0c;部署deployment、service、ingress等等 2、helm的作用&#xff1a;通过打包的方式&#xff0c;deployment、service、ingress这些打包在一块&#xff0c;一键部署服务、类似于yum功能 3、helm&#xff1a;官方提供的一种类似于仓库的功能&#…
阅读更多...
时间轮设计

时间轮设计

目录 基本概念 函数定义 函数实现与测试 测试1结果如下 测试2结果如下 基本概念 时间轮 是一种 实现延迟功能&#xff08;定时器&#xff09; 的 巧妙算法。如果一个系统存在大量的任务调度&#xff0c;时间轮可以高效的利用线程资源来进行批量化调度。把大批量的调度任务…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023