Spark Spark的基础环境 Day02

Posted Maynor大数据

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark Spark的基础环境 Day02相关的知识,希望对你有一定的参考价值。

Spark Day03:Spark 基础环境

02-[了解]-今日课程内容提纲

主要讲解2个方面内容:Spark on YARN集群和RDD 是什么

1、Spark on YARN
	将Spark应用程序,提交运行到YARN集群上,企业中绝大多数运行模式,必须掌握
	- 如何配置
	- 提交应用运行
	- Spark应用运行在集群上2种Deploy-Mode
	- yarn-client模式
	- yarn-cluster模式

2、RDD是什么
	RDD,弹性分布式数据集,抽象概念,相当于集合,比如列表List,分布式集合,存储海量数据
	引入RDD数据结构
	RDD 官方定义,从文档和源码
	RDD 5大特性(面试必问)
	词频统计WordCount查看RDD有哪些
	
	RDD创建方式,如何将数据封装到RDD集合中,2种方式
	创建RDD时,如何处理小文件(面试)

03-[掌握]-Spark on YARN之属性配置和服务启动

将Spark Application提交运行到YARN集群上,至关重要,企业中大多数都是运行在YANR上

文档:http://spark.apache.org/docs/2.4.5/running-on-yarn.html

​ 当Spark Application运行到YARN上时,在提交应用时指定master为yarn即可,同时需要告知YARN集群配置信息(比如ResourceManager地址信息),此外需要监控Spark Application,配置历史服务器相关属性

在实际项目中,只需要配置:6.1.1 至 6.1.4即可,由于在虚拟机上测试,所以配置6.1.5解除资源检查限制。

04-[掌握]-Spark on YARN之提交应用

先将圆周率PI程序提交运行在YARN上,命令如下:

SPARK_HOME=/export/server/spark
$SPARK_HOME/bin/spark-submit \\
--master yarn \\
--class org.apache.spark.examples.SparkPi \\
$SPARK_HOME/examples/jars/spark-examples_2.11-2.4.5.jar \\
10

运行完成在YARN 监控页面截图如下

设置资源信息,提交运行WordCount程序至YARN上,命令如下:

SPARK_HOME=/export/server/spark
$SPARK_HOME/bin/spark-submit \\
--master yarn \\
--driver-memory 512m \\
--executor-memory 512m \\
--executor-cores 1 \\
--num-executors 2 \\
--queue default \\
--class cn.itcast.spark.submit.SparkSubmit \\
hdfs://node1.itcast.cn:8020/spark/apps/spark-day02_2.11-1.0.0.jar \\
/datas/wordcount.data /datas/swcy-output

当WordCount应用运行YARN上完成以后,从8080 WEB UI页面点击应用历史服务连接,查看应用运行状态信息。

05-[掌握]-DeployMode两种模式区别

Spark Application提交运行时部署模式Deploy Mode,表示的是Driver Program运行的地方,要么是提交应用的Client:client,要么是集群中从节点(Standalone:Worker,YARN:NodeManager):cluster

  • client 模式

​ 默认DeployMode为Client,表示应用Driver Program运行在提交应用Client主机上(启动JVM Process进程),示意图如下:

假设运行圆周率PI程序,采用client模式,命令如下:

SPARK_HOME=/export/server/spark
$SPARK_HOME/bin/spark-submit \\
--master spark://node1.itcast.cn:7077,node2.itcast.cn:7077 \\
--deploy-mode client \\
--driver-memory 512m \\
--executor-memory 512m \\
--executor-cores 1 \\
--total-executor-cores 2 \\
--class org.apache.spark.examples.SparkPi \\
$SPARK_HOME/examples/jars/spark-examples_2.11-2.4.5.jar \\
10
  • cluster 模式

如果采用cluster模式运行应用,应用Driver Program运行在集群从节点Worker某台机器上。

假设运行圆周率PI程序,采用cluster模式,命令如下:

SPARK_HOME=/export/server/spark
$SPARK_HOME/bin/spark-submit \\
--master spark://node1.itcast.cn:7077,node2.itcast.cn:7077 \\
--deploy-mode cluster \\
--supervise \\
--driver-memory 512m \\
--executor-memory 512m \\
--executor-cores 1 \\
--total-executor-cores 2 \\
--class org.apache.spark.examples.SparkPi \\
$SPARK_HOME/examples/jars/spark-examples_2.11-2.4.5.jar \\
10

06-[掌握]-Spark on YARN之YARN Client 模式

当应用运行YARN上时,有2部分组成:

  • AppMaster,应用管理者,申请资源和调度Job执行
  • Process,运行在NodeManager上进程,运行Task任务

Spark 应用运行集群上时,也有2部分组成:

  • Driver Program,应用管理者,申请资源运行Executors和调度Job执行
  • Executors,运行JVM进程,其中执行Task任务和缓存数据

当Spark应用运行在YARN集群上时,运行架构是什么样子的呢????

  • YARN Client 模式

当Spark 运行在YARN集群时,采用client DeployMode时,有如下三个进程:

  • AppMaster,申请资源,运行Executors
  • Driver Program,调度Job执行和监控
  • Executors,运行JVM进程,其中执行Task任务和缓存数据
  • YARN Cluster 模式

当Spark 运行在YARN集群时,采用clusterDeployMode时,有如下2个进程:

  • Driver Program(AppMaster),既进行资源申请,又进行Job调度
  • Executors,运行JVM进程,其中执行Task任务和缓存数据

所以Spark Application运行在YARN上时,采用不同DeployMode时架构不一样,企业实际生产环境还是以cluster模式为主,client模式用于开发测试,两者的区别面试中常问。


在YARN Client模式下,Driver在任务提交的本地机器上运行,示意图如下:

采用yarn-client方式运行词频统计WordCount程序

/export/server/spark/bin/spark-submit \\
--master yarn \\
--deploy-mode client \\
--driver-memory 512m \\
--executor-memory 512m \\
--executor-cores 1 \\
--num-executors 2 \\
--queue default \\
--class cn.itcast.spark.submit.SparkSubmit \\
hdfs://node1.itcast.cn:8020/spark/apps/spark-day02_2.11-1.0.0.jar \\
/datas/wordcount.data /datas/swcy-client

07-[掌握]-Spark on YARN之YARN Cluster模式

​ 在YARN Cluster模式下,Driver运行在NodeManager Contanier中,此时Driver与AppMaster合为一体,示意图如下:

以运行词频统计WordCount程序为例,提交命令如下:

/export/server/spark/bin/spark-submit \\
--master yarn \\
--deploy-mode cluster \\
--driver-memory 512m \\
--executor-memory 512m \\
--executor-cores 1 \\
--num-executors 2 \\
--queue default \\
--class cn.itcast.spark.submit.SparkSubmit \\
hdfs://node1.itcast.cn:8020/spark/apps/spark-day02_2.11-1.0.0.jar \\
/datas/wordcount.data /datas/swcy-cluster

08-[理解]-Spark 应用MAIN函数代码执行

Spark Application应用程序运行时,无论client还是cluster部署模式DeployMode,当DriverProgram和Executors启动完成以后,就要开始执行应用程序中MAIN函数的代码,以词频统计WordCount程序为例剖析讲解。

上述图片中,A、B都是在Executor中执行,原因在于对RDD数据操作的,针对C来说,如果没有返回值时,在Executor中执行,有返回值,比如调用count、first等函数时,在Driver中执行的。

09-[了解]-RDD 概念之引入说明

​ 对于大量的数据,Spark 在内部保存计算的时候,都是用一种叫做弹性分布式数据集(ResilientDistributed Datasets,RDD)的数据结构来保存的,所有的运算以及操作都建立在 RDD 数据结构的基础之上

在Spark框架中,将数据封装到集合中:RDD,如果要处理数据,调用集合RDD中函数即可。

也就是说RDD设计的核心点为:

文档:http://spark.apache.org/docs/2.4.5/rdd-programming-guide.html

10-[掌握]-RDD 概念之官方定义

​ RDD(Resilient Distributed Dataset)叫做弹性分布式数据集,是Spark中最基本的数据抽象,代表一个不可变可分区、里面的元素可并行计算的集合。

拆分核心要点三个方面:

​ 可以认为RDD是分布式的列表List或数组Array,抽象的数据结构,RDD是一个抽象类AbstractClass和泛型Generic Type

RDD弹性分布式数据集核心点示意图如下:

11-[掌握]-RDD 概念之5大特性剖析

RDD 数据结构内部有五个特性(摘录RDD 源码):前3个特性,必须包含的;后2个特性,可选的。

在RDD类中,对应一个方法:

​ RDD 是一个数据集的表示,不仅表示了数据集,还表示了这个数据集从哪来、如何计算,主要属性包括五个方面(必须牢记,通过编码加深理解,面试常问):

12-[掌握]-RDD 概念之词频统计WordCount中RDD

以词频统计WordCount程序为例,查看整个Job中各个RDD类型及依赖关系,WordCount程序代码如下:

运行程序结束后,查看WEB UI监控页面,此Job(RDD调用foreach触发)执行DAG图:

13-[掌握]-RDD 创建的两种方式

​ 如何将数据封装到RDD集合中,主要有两种方式:并行化本地集合(Driver Program中)和引用加载外部存储系统(如HDFS、Hive、HBase、Kafka、Elasticsearch等)数据集。

  • 并行化集合

    • 由一个已经存在的 Scala 集合创建,集合并行化,集合必须时Seq本身或者子类对象。

    • 只能将Scala中Seq对象或者子类对象,并行化RDD

      package cn.itcast.spark.source
      
      import org.apache.spark.rdd.RDD
      import org.apache.spark.SparkConf, SparkContext
      
      /**
       * Spark 采用并行化的方式构建Scala集合Seq中的数据为RDD
      	 * - 将Scala集合转换为RDD
      	 *      sc.parallelize(seq)
      	 * - 将RDD转换为Scala中集合
      		 * rdd.collect() -> Array
      		 * rdd.collectAsMap() - Map,要求RDD数据类型为二元组
       */
      object _01SparkParallelizeTest 
      	
      	def main(args: Array[String]): Unit = 
      		
      		val sc: SparkContext = 
      			// sparkConf对象
      			val sparkConf = new SparkConf()
      				// _01SparkParallelizeTest$  ->(.stripSuffix("$"))   ->  _01SparkParallelizeTest
          			.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
          			.setMaster("local[2]")
      			// sc 实例对象
      			SparkContext.getOrCreate(sparkConf)
      		
      		
      		// TODO: 1、Scala中集合Seq序列存储数据
      		val linesSeq: Seq[String] = Seq(
      			"hadoop scala hive spark scala sql sql", //
      			"hadoop scala spark hdfs hive spark", //
      			"spark hdfs spark hdfs scala hive spark" //
      		)
      		
      		// TODO: 2、并行化集合
      		val inputRDD: RDD[String] = sc.parallelize(linesSeq, numSlices = 2)
      		
      		// TODO: 3、词频统计
      		val resultRDD = inputRDD
      			.flatMap(line => line.split("\\\\s+"))
      			.map(word => (word, 1))
      			.reduceByKey((tmp, item) => tmp + item)
      		
      		// TODO: 4、输出结果
      		resultRDD.foreach(println)
      		
      		// 应用结束,关闭资源
      		sc.stop()
      	
      	
      	
      
      
  • 外部存储系统

实际使用最多的方法:textFile,读取HDFS或LocalFS上文本文件,指定文件路径和RDD分区数目。

实际项目中如果从HDFS读取海量数据,应用运行在YARN上,默认情况下,RDD分区数目等于HDFS上Block块数目。

14-[掌握]-创建RDD时小文件读取

​ 在实际项目中,有时往往处理的数据文件属于小文件(每个文件数据数据量很小,比如KB,几十MB等),文件数量又很大,如果一个个文件读取为RDD的一个个分区,计算数据时很耗时性能低下,使用SparkContext中提供:wholeTextFiles类,专门读取小文件数据。

范例演示:读取100个小文件数据,每个文件大小小于1MB,设置RDD分区数目为2。

package cn.itcast.spark.source

import org.apache.spark.rdd.RDD
import org.apache.spark.SparkConf, SparkContext

/**
 * 采用SparkContext#wholeTextFiles()方法读取小文件
 */
object _02SparkWholeTextFileTest 
	
	def main(args: Array[String]): Unit = 
		val sc: SparkContext = 
			// sparkConf对象
			val sparkConf = new SparkConf()
				// _01SparkParallelizeTest$  ->(.stripSuffix("$"))   ->  _01SparkParallelizeTest
				.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
				.setMaster("local[2]")
			// sc 实例对象
			SparkContext.getOrCreate(sparkConf)
		
		
		/*
		  def wholeTextFiles(
		      path: String,
		      minPartitions: Int = defaultMinPartitions
		  ): RDD[(String, String)]
		  Key: 每个小文件名称路径
		  Value:每个小文件的内容
		 */
		val inputRDD: RDD[(String, String)] = sc.wholeTextFiles("datas/ratings100", minPartitions = 2)
		
		println(s"RDD 分区数目 = $inputRDD.getNumPartitions")
		
		inputRDD.take(2).foreach(tuple => println(tuple))
		
		// 应用结束,关闭资源
		sc.stop()
		
	
	


以上是关于Spark Spark的基础环境 Day02的主要内容,如果未能解决你的问题,请参考以下文章

Note_Spark_Day02:Standalone集群模式和使用IDEA开发应用程序

Spark基础学习笔记02:搭建Spark环境

Note_Spark_Day01:Spark 框架概述和Spark 快速入门

spark day02+day03

2022年Spark基础学习笔记

Spark集群搭建