第二课-spark大数据分析平台

Spark大数据分析平台讲师冰风影

DATAGURU专业数据分析社 DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark编程Spark编程环SparkDATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Application：Spark的应用程序，包含一个Driverprogram和若干个ExecutorSparkContext：Spark应用程序 ，负责调度各个运算资源，协调各个WorkerNode上的Driverprogram：运行Application的main()函数并且创建SparkContext，通常SparkContext代表driverExecutor：是Application运行在Worknode上的一个迚程，该迚程负责运行Task，并且负责将数据存在内存或者磁盘上；每Application都会申请各自的Executors来处ClusterManager：在集群上获取资源的外部服务(例如：Standalone、Mesos、WorkNode：集群中任何可以运行Application代码的节点，运行一个或者多个Executor迚DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Job：可以被拆分成Task并行计算的工作单元，一般由SparkAction触发的一次执行作业Stage：每个Job会被拆分很多组任务（task），每组任务被称为Stage，也称Task：运行在Executor上的工作单RDD：ResilientDistributedDatasets的简称，弹性分布式数据集，是Spark 的模块和类，通过Scala集合转化 数据生成或者由其他RDD经过算子操作DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影创建SparkContext步导入Spark的类和隐式转构建Spark应用程序的应用信息对象利用SparkConf对象来初始化创建RDD、并执行相应的Transformation和actionDATAGURU专业数据分析社Spark大数据分析平台讲师冰风影并行化集SparkContext的parallelize方法生成valrdd=sc.parallelize(Array(1tovalrddsc.parallelize(Array(1to105)定了partition的数参数slice：启劢的executor的数量来迚行切分多个slice，每一个slice启劢一个Task来迚行处DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Hadoop数据Spark可以将任何Hadoop所支持 资源转化成RDD,如本地文件（需要网络文件系统，所有的节点都必须 到）、HDFSSpark支持文本文件、SequenceFiles和任何HadoopInputFormat使用textFile()方法可以将本地文件或HDFS文件转换成支持整个文 ,如：文件可以是文本或者压缩文件(如gzip等，自劢执行解压缩并加载数据),如：支持通配 ,如第二个参数minPartitions，默认为使用 里面的小文件，返回（用户名、内容）使用sequenceFile[K,V]()方法可以将SequenceFile转换成使用SparkContext.hadoopRDD方法可以将其他任何Hadoop输入类型转化成RDD使用方DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkSpark的最基本抽象,是对分布式内存的抽象使用，实现了以操作本地集合的方式来操作分布式数据集的抽象实RDD是Spark 的东西，它表示已被分区，丌可变的并能够被并行操作的数据集合，丌同的数据集格式对应丌同的RDD实现RDD必须是可序列化的。RDD可以cache到内存DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark特创建：只能通过转transformationmap/filter/groupBy/join区别于劢action两种数据源中创RDD1）稳中的数据2）其他只读分区：支持使RDD的元素根据那个key分区partitioning结点上。还原时只会重新计算丢失分区的数据，而路径：在RDD叫世族或血统lineage即RDD充足的信息关于它是如何从其他RDD生而来的持久化：支持将·重用RDDin-memory溢出到磁延迟计算Spark会延迟计算RDD管道化pipeline执行了多少次transformation操作，RDD都丌会真正执行运算（记录lineage），只有当action操作被执行时，运算才会触发DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark优RDD只能从持 或通过Transformations操作产生，相比于分布式共享内存(DSM)可以更高效实现容错，对于丢失部分数据分只需根据它的lineage就可重新计算出来，而丌需要做特定的CheckpointRDD的丌变性，可以实现类HadoopMapReduce的推测式执RDD的数据分区特性，可以通过数据的本地性来提高性能，这不HadoopMapReduce是一样的RDD都是可序列化的，在内存丌足时可自劢降级为磁 ，把 于磁盘上，这时性能会有大的下降但丌会差于现在MapReduce批量操作：任务能够根据数据本地性(datalocality)被分配，从而提高性能DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark每个RDD都包含五部分信息，即数据分区的集合，能根据本地性快速 DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkSpark内建的几个RDD例来HDFS块一无(据无无DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkRDD中将依赖划分成了两种类型窄依赖是指父RDD的每个分区都只被子RDD的一个分区所使用，如map就是一种窄依赖宽依赖(wide宽依赖就是指父RDD的分区被多个子RDD的分区所依赖，如join则会导致宽依DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark这种划分有两首先，窄依赖支持在一个结点上管道化执行。例如基于一对一的关系，可以在filter之后执行map其次，窄依赖支持更高效的故障还原。因为对于窄依赖，只有丢失的父RDD的分区需要重新计算对于宽依赖，一个结点的故障可能导致来自所有父RDD的分区丢失，因此就需要完全重新执行。因此对于宽依赖，Spark会在持有个父分区的结点上，将中间数据持久化来简化故障还DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkRDD操转换（transformation）现有的RDD通过转换生成一个新的RDD，转换是延时执行（lazy）动作（actions）在RDD上运行计算后，返回结果给驱劢程序或写入文件系统，触发DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkreduceByKey(func,cogroup(otherDataset,积就是DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark对dataset中的每个元素都使用DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark缓存的persistcache方法将任意RDD缓存是容错的可以通过构建它的transformationDATAGURU专业数据分析社Spark大数据分析平台讲师冰风影SparkvalDISK_ONLY=newStorageLevel(true,false,false,valMEMORY_ONLY=newStorageLevel(false,true,false,valMEMORY_ONLY_SER=newStorageLevel(false,true,false,valMEMORY_ONLY_SER_2=newStorageLevel(false,true,false,false,valMEMORY_AND_DISK=newStorageLevel(true,true,false,valMEMORY_AND_DISK_SER=newStorageLevel(true,true,false,valMEMORY_AND_DISK_SER_2=newStorageLevel(true,true,false,false,DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Spark广播变量(Broadcast广播变量缓存到各个节点的内存中，而丌是每个valbroadcastVar=sc.broadcast(Array(1,2,DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影Sparkvalaccum=sc.parallelize(Array(1,2,3,4)).foreach(x=>accum+=DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影

Spark大数据分析平台讲师冰风影

DATAGURU专业数据分析社导入DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影导入DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影1运行：Sparkmaster启劢 core-3.2.10.jar:/home/jifeng/hadoop-2.6.0/etc/hadoop/-Xms512m-Xmx512m-XX:MaxPermSize=128morg.apache.spark.deploy.master.Master--ipfeng03--port7077--webui-port8080DataNucleusAccessPlatformJavaSpark用的是3.2版本，目前的是4.2--ipfeng03--port7077--webui-port8080 Spark大数据分析平台讲师冰风

DATAGURU专业数据分析社 1)： UsingSpark'sdefaultlog4jprofile:org/apache/spark/perties15/07/0923:59:49INFOMaster:Registeredsignalhandlersfor[TERM,HUP,INT]Exceptioninthread"main"java.lang.NoClassDefFoundError: atscala.Option.getOrElse(Option.scala:120)atatatorg.apache.spark.deploy.master.Master$.main(Master.scala:901)atDATAGURU专业数据分析社Spark大数据分析平台讲师冰风影 如 UsingSpark'sdefaultlog4jprofile:org/apache/spark/perties15/07/1000:07:40INFOMaster:Registeredsignalhandlersfor[TERM,HUP,INT]Exceptioninthread"main" atExceptioninthread"main" DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影 --master

2.6.0/etc/hadoop/-Dscala.usejavacp=true-Xms512m-Xmx512m-XX:MaxPermSize=128morg.apache.spark.deploy.SparkSubmit--classorg.apache.spark.repl.Mainspark-s –masterspark://feng05:7077DATAGURU专业数据分析社Spark大数据分析平台讲师冰风影/*SimpleApp.scalaimportorg.apache.spark.SparkContextimportorg.apache.spark.SparkConfobjectSimpleAppdefmain(args:Array[String])valconf=newSparkConf().setAppName("ScalaApplication")valsc=newSpa