当涂 微网站开发公司,贵阳网站建设推广公司,如何做网站美工,萍乡网站建设行吗第 1 章#xff1a;Spark概述
1.1 什么是spark
回顾#xff1a;hadoop主要解决#xff0c;海量数据的存储和海量数据的分析计算。 spark是一种基于内存的快速、通用、可扩展的大数据分析计算引擎。
1.2 hadoop与spark历史
hadoop的yarn框架比spark框架诞生的晚#xff…第 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安装包
[atguiguhadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguiguhadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-local
2官方求pi案例
[atguiguhadoop102 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所用参数
[atguiguhadoop102 spark-local]$ bin/spark-submit–class表示要执行程序的主类 –master local[2]“ 1local没有指定线程数则所有计算都运行在一个线程当中没有任何并行计算。 2local[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)
3local[*]默认模式。自动帮你按照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准备文件
[atguiguhadoop102 spark-local]$ mkdir input在Input下创建2个文件1.txt和2.txt并输入一下内容
hello atguigu
hello spark
2启动spark-shell
[atguiguhadoop102 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 Sparks 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程序入口masterlocal[*]表示本地模式运行。 3再开启一个hadoop102远程连接窗口发现了一个sparksubmit进程
[atguiguhadoop102 spark-local]$ jps
3627 SparkSubmit
4047 Jps
运行任务方式说明spark-submit是将jar上传到集群执行spark任务spark-shell相当于命令行工具本身也是一个application。 4登录hadoop102:4040查看程序运行情况
说明本地模式下默认的调度器为fifo。 5运行workcount程序
scalasc.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自带的资源调度引擎构建一个由masterworker构成的spark集群spark运行在集群种。 这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群不需要借助hadoop的yarn和mesos等其它框架。
2.3.1 master和worker集群资源管理
masterspark特有资源调度系统的leader。掌管着整个集群的资源信息类似于yarn框架中的resourcemanager。 workerspark特有资源调度系统的slave有多个。每个slave掌管着所在节点的资源信息类似于yarn框架中的nodemanager。
master和worker是spark的守护进程、集群资源管理者即spark在特定模式standalone下正常运行必须要有的后台常驻进程。
2.3.2 driber和executor任务的管理者 driver和executor是临时程序当有具体任务提交到spark集群才会开启的程序。standalone模式是spark自带的资源调度引擎构建一个由masterworker构成spark集群spark运行在集群中。 这个要和hadoop中的standalone区别开来。这里的standalone是指只用spark来搭建一个集群不需要借助hadoop的yarn和mesos等其它框架。
2.3.2 安装使用
1、集群规划
2、再解压一份spark安装包并修改解压后的文件夹名称为spark-standalone
[atguiguhadoop102 sorfware]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/
[atguiguhadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2 spark-standalone
3、进入spark的配置文件/opt/module/spark-standalone/conf
[atguiguhadoop102 spark-standalone]$ cd conf4、修改slave文件添加work节点
atguiguhadoop102 conf]$ mv slaves.template slaves
[atguiguhadoop102 conf]$ vim slaves
hadoop102
hadoop103
hadoop104
5、修改spark-env.sh文件添加master节点
[atguiguhadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguiguhadoop102 conf]$ vim spark-env.shSPARK_MASTER_HOSThadoop102
SPARK_MASTER_PORT7077
6、分发spark-standalone包
[atguiguhadoop102 module]$ xsync spark-standalone/7、启动spark集群
[atguiguhadoop102 module]$ xsync spark-standalone/查看三台服务器运行进程xcall.sh是以前数仓项目里面讲的脚本
[atguiguhadoop102 spark-standalone]$ xcall.sh jps
atguiguhadoop102
3238 Worker
3163 Master
atguiguhadoop103
2908 Worker
atguiguhadoop104
2978 Worker
注意如果遇见”Java_home not set“异常可以在sbin目录下的spark-config.sh文件中加入如下配置
export JAVA_HOMEXXXX8、网页查看hadoop102:8080 9、官方求pi案例
[atguiguhadoop102 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. 8080master的webui 4040application的webui的端口号
2.3.3 参数说明
1、配置executor可用内存为2G使用cpu核数为2个
[atguiguhadoop102 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名称
[atguiguhadoop102 conf]$ mv spark-defaults.conf.template spark-defaults.conf2、修改spark-default.conf文件配置日志存储路径
[atguiguhadoop102 conf]$ vim spark-defaults.conf
spark.eventLog.enabled true
spark.eventLog.dir hdfs://hadoop102:8020/directory
注意需要启动hdaoop集群hdfs上的目录需要提前存在
[atguiguhadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguiguhadoop102 hadoop-3.1.3]$ hadoop fs -mkdir /directory
3、修改spark-env.sh文件添加如下配置
[atguiguhadoop102 conf]$ vim spark-env.shexport SPARK_HISTORY_OPTS
-Dspark.history.ui.port18080
-Dspark.history.fs.logDirectoryhdfs://hadoop102:8020/directory
-Dspark.history.retainedApplications30
1参数1含义webui访问的端口号为18080 2参数2含义指定历史服务器日志存储路径读 3参数3含义指定保存application历史记录的个数如果超过这个值旧的应用程序信息将被删除这个是内存中的应用数而不是页面上的显示的应用数 4、分发配置文件
[atguiguhadoop102 conf]$ xsync spark-defaults.conf spark-env.sh5、启动历史服务
[atguiguhadoop102 spark-standalone]$
sbin/start-history-server.sh
6、再次执行任务
[atguiguhadoop102 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停止集群
[atguiguhadoop102 spark-standalone]$ sbin/stop-all.sh2zookeeper正常安装并启动基于以前讲的数仓项目脚本
[atguiguhadoop102 zookeeper-3.4.10]$ zk.sh start3修改spark-env.sh文件添加如下配置
[atguiguhadoop102 conf]$ vim spark-env.sh#注释掉如下内容
#SPARK_MASTER_HOSThadoop102
#SPARK_MASTER_PORT7077#添加上如下内容。配置由Zookeeper管理Master在Zookeeper节点中自动创建/spark目录用于管理
export SPARK_DAEMON_JAVA_OPTS
-Dspark.deploy.recoveryModeZOOKEEPER
-Dspark.deploy.zookeeper.urlhadoop102,hadoop103,hadoop104
-Dspark.deploy.zookeeper.dir/spark#添加如下代码
#Zookeeper3.5的AdminServer默认端口是8080和Spark的WebUI冲突
export SPARK_MASTER_WEBUI_PORT8989
4分发配置文件
[atguiguhadoop102 conf]$ xsync spark-env.sh5在hadoop102上启动全部节点
[atguiguhadoop102 spark-standalone]$ sbin/start-all.sh6在hadoop103上单独启动master节点
[atguiguhadoop103 spark-standalone]$ sbin/start-master.sh7在启动一个hadoop102窗口将/opt/module/spark-local/input数据上传到hadoop集群的/input目录
[atguiguhadoop102 spark-standalone]$ hadoop fs -put /opt/module/spark-local/input/ /input8spark ha集群访问
[atguiguhadoop102 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程序
scalasc.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进程
[atguiguhadoop102 ~]$ jps
5506 Worker
5394 Master
5731 SparkSubmit
4869 QuorumPeerMain
5991 Jps
5831 CoarseGrainedExecutorBackend
2kill掉hadoop102的master进程页面中观察http://hadoop103:8080/的状态是否切换为active
[atguiguhadoop102 ~]$ kill -9 53943再启动hadoop102的master进程
[atguiguhadoop102 spark-standalone]$ sbin/start-master.sh2.3.5 运行流程
spark由standalone-client核standalone-cluster两种模式主要区别在于driver程序的运行节点。 1、客户端模式
[atguiguhadoop102 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、集群模式
[atguiguhadoop102 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集群
[atguiguhadoop102 spark-standalone]$ sbin/stop-all.sh
[atguiguhadoop102 spark-standalone]$ zk.sh stop
[atguiguhadoop103 spark-standalone]$ sbin/stop-master.sh
2、为了防止和standalone模式冲突再单独解压一份spark
[atguiguhadoop102 software]$ tar -zxvf spark-3.1.3-bin-hadoop3.2.tgz -C /opt/module/3、进入到/opt/module目录修改spark-~名称为spark-yarn
[atguiguhadoop102 module]$ mv spark-3.1.3-bin-hadoop3.2/ spark-yarn4、修改hadoop配置文件/opt/module/~/yarn-site.xml添加如下内容 因为测试环境虚拟机内存较少防止执行过程进行倍意外杀死做如下处理
[atguiguhadoop102 hadoop]$ vim yarn-site.xml
!--是否启动一个线程检查每个任务正使用的物理内存量如果任务超出分配值则直接将其杀掉默认是true --
propertynameyarn.nodemanager.pmem-check-enabled/namevaluefalse/value
/property!--是否启动一个线程检查每个任务正使用的虚拟内存量如果任务超出分配值则直接将其杀掉默认是true --
propertynameyarn.nodemanager.vmem-check-enabled/namevaluefalse/value
/property
5、分发配置文件
[atguiguhadoop102 conf]$ xsync /opt/module/hadoop-3.1.3/etc/hadoop/yarn-site.xml6、修改/opt/~/spark-env.sh添加yarn_conf_dir配置保证后续运行任务的路径都编程集群路径
[atguiguhadoop102 conf]$ mv spark-env.sh.template spark-env.sh
[atguiguhadoop102 conf]$ vim spark-env.shYARN_CONF_DIR/opt/module/hadoop-3.1.3/etc/hadoop
7、启动hdfs以及yarn集群
[atguiguhadoop102 hadoop-3.1.3]$ sbin/start-dfs.sh
[atguiguhadoop103 hadoop-3.1.3]$ sbin/start-yarn.sh
8、执行一个程序
[atguiguhadoop102 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历史服务器关联路径 目的点击yarn8088上spark任务的history按钮进入的是spark历史服务器18080而不再是yarn历史服务器19888 1、修改配置文件/opt/module/~/spark-defaults.conf 添加如下内容
spark.yarn.historyserver.addresshadoop102:18080
spark.history.ui.port180802、重启spark历史服务
[atguiguhadoop102 spark-yarn]$ sbin/stop-history-server.sh [atguiguhadoop102 spark-yarn]$ sbin/start-history-server.sh
3、提交任务到yarn执行
[atguiguhadoop102 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-clientdriver程序运行在客户端适用于交互、调试希望立即看到app的输出 yarn-clusterdriver程序运行在由resourcemanager启动的appmaster适用于生产环境 1、客户端模式默认
[atguiguhadoop102 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、集群模式
[atguiguhadoop102 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按钮跳转到历史详情页面
2http://hadoop102:18080点击executors-点击driver的stdout
注意如果在yarn日志端无法查看到具体的日志则在yarn-site.xml中添加如下配置并启动yarn历史服务器
propertynameyarn.log.server.url/namevaluehttp://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的8032rm和nm的内部通信端口 3、spark standalone模式master web端口号8080类似于hadoop yarn任务运行情况查看端口号8088yarn模式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上面必须要有
dependenciesdependencygroupIdorg.apache.spark/groupIdartifactIdspark-core_2.12/artifactIdversion3.1.3/version/dependency
/dependenciesbuildfinalNameWordCount/finalNamepluginsplugingroupIdnet.alchim31.maven/groupIdartifactIdscala-maven-plugin/artifactIdversion3.4.6/versionexecutionsexecutiongoalsgoalcompile/goalgoaltestCompile/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 》11 (atguigu,2)val wordToSumRdd: RDD[(String, Int)] wordToOneRdd.reduceByKey(__)//7.将统计结果采集到控制台打印wordToSumRdd.collect().foreach(println)//8.关闭连接sc.stop()}
}
2、调试流程
spark程序运行过程中会打印大量的执行日志为了能够更好的查看程序的执行结果可以在项目的resources目录中创建log4j.properties文件并添加日志配置文件
log4j.rootCategoryERROR, console
log4j.appender.consoleorg.apache.log4j.ConsoleAppender
log4j.appender.console.targetSystem.err
log4j.appender.console.layoutorg.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 loggers log level, so that
# the user can have different defaults for the shell and regular Spark apps.
log4j.logger.org.apache.spark.repl.MainERROR# Settings to quiet third party logs that are too verbose
log4j.logger.org.spark_project.jettyERROR
log4j.logger.org.spark_project.jetty.util.component.AbstractLifeCycleERROR
log4j.logger.org.apache.spark.repl.SparkIMain$exprTyperERROR
log4j.logger.org.apache.spark.repl.SparkILoop$SparkILoopInterpreterERROR
log4j.logger.org.apache.parquetERROR
log4j.logger.parquetERROR# SPARK-9183: Settings to avoid annoying messages when looking up nonexistent UDFs in SparkSQL with Hive support
log4j.logger.org.apache.hadoop.hive.metastore.RetryingHMSHandlerFATAL
log4j.logger.org.apache.hadoop.hive.ql.exec.FunctionRegistryERROR
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
[atguiguhadoop102 spark-yarn]$ hadoop fs -mkdir /input4上传输入文件到/input路径
[atguiguhadoop102 spark-yarn]$ hadoop fs -put /opt/module/spark-local/input/1.txt /input5执行任务
[atguiguhadoop102 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查看运行结果
[atguiguhadoop102 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变量
