spark杰普软件科技有限公司

1、杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：1Briup Training What is Spark Spark简介( Spark4个特点与发展) BDAS系统 Why Spark Spark与MapReduce的差异 Spark的适用场景 Spark How Spark案例安装Spark Spark常用命令

2、Spark架构组件和工作机制Spark运行原理Spark-Core内容内容Spark-SqlSpark-Streaming内容2杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：3Briup Training Spark是加州大学伯克利分校AMP开发通用内存并行计算框架。 Apache Spark is a fast

3、 and general engine for large-scale data processing. Spark是基于内存计算的大数据分布式计算框架。Spark基于内存计算，提用户将Spark部署在大供可交互的方式，提供近实时的处理方式，量廉价硬件之上，形成集群，提高处理速度，同时也保证了集群高容错性和高可伸缩性。4Briup Training 2009年伯克利分校开始使用Scala语言编写Spark源码，2010年开放源代码，2013年6月进入Apache孵化项目，2014年2月成为Apache顶级项目，2014年5月底发布Spark1.0.0,到目前为止版本更新为Spark2.4.3。

4、5Briup Training 演进时间表：2009年由 Berkeleys AMPLab 开始编写最初的源代码2010年开放源代码2013年6月进入Apache孵化器项目2014年2月成为Apache的顶级项目（82014年5月底Spark1.0.0发布2014年9月Spark1.1.0发布.2016年3月Spark1.6.1发布2016年7月Spark2.0发布2016年10月Spark2.0.1发布2016年11月Spark1.6.3发布2016年11月Spark2.0.2发布.2017年10月Spark2.1.2发布2017年12月Spark2.2.1发布2018年2月Spark2.3

5、.0发布时间）6Briup Training 在Spark官网上介绍，它具有运行速度快、易用性好、通用性强和随处运行等特点。7Briup Training 运行速度快 Spark拥有DAG执行引擎，支持在内存中对数据进行迭代计算。提供的数据，速度是Hadoop MapReduce的10倍以上，如果数据表明，如果数据由磁盘，速度可以高达100多倍。从内存中8Briup Training 易用性 支持4种语言的API：scala、java、python、R。特别是Scala是一种高效、可拓展的语言，能够用简洁的代码处理较为复杂的处理工作。 val text_file=sc.textFile(“ h

6、dfs:/.” ) text_file.flatMap(_.split(“ text_file.flatMap(_.split(“” ).map(x = (x,1).reduceByKey(_+_)” ).countByValue();9Briup Training scala val textFile = spark.read.textFile(“ README.md) val wordCounts = textFile.flatMap(line = line.split( ).groupByKey(identity).count()1Briup Training 通用性强 Spark圈即B

7、DAS（伯克利数据分析栈）包含了Spark Core、Spark SQL、Spark Streaming、MLLib和GraphX等组件，这些组件分别处理Spark Core提供内存计算框架、SparkStreaming的实时处理应用、Spark SQL的即席、MLlib或MLbase的学习和GraphX的图处理，它们都是由AMP提供，能够无缝的集成并提供一站式解决平台。1Briup Training 随处运行 Spark具有很强的适应性，能够HDFS、HBase、Cassandra S3和Techyon为持久层读写原生数据，能够以Mesos、YARN和自身携带的Standalone作为管理器

8、调度job，来完成Spark应用程序的计算。12Briup Training Spark圈也称为BDAS（伯克利数据分析栈），是伯克利APMLab打造的，力图在算法（Algorithms）、（Machines）、人（People）之间通过大规模集成来展现大数据应用的一个平台。伯克利AMPLab运用大数据、云计算、通信等各种以及各种灵活的技术方案，对海量不透明的数据进行甄别并转化为有用的信息，以供人们更好的理解世界。该圈已经涉及到学习、数据挖掘、数据库、信息检索、自然语言处理和语音识别等多个领域。13Briup Training Spark圈以Spark Core为，从HDFS、HBase和Am

9、azon S3等持久层数据，以MESOS、YARN和自身携带的Standalone为管理器调度Job完成Spark应用程序的计算。 这些应用程序可以来自于不同的组件，如SparkShell/Spark Submit的批处理、Spark Streaming的实时处理应用、Spark SQL、BlinkDB的权衡、MLlib/MLbase的学习、GraphX的图处的即席理和SparkR的数学计算等等。14Briup Training15杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软

10、件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：16Briup Training Spark是在借鉴了MapReduce之上发展而来的，继承了其分布式并行计算的优点并改进了MapReduce明显的缺陷，具体如下：基于内存计算容错性高通用性好17Briup Training 首先，Spar k把中间数据放到内存中，迭代运算效率高。MapReduce中计算结果需要落地，保存到磁盘上，这样势必会影响整体速度，而Spar k支持DAG图的分布式并行计算的编程框架，减少了迭代过程中数据

11、的落地，提高了处理效率。 其次，Spar k容错性高。Spar k引进了弹性分布式数据集RDD (Resilient Distr ibuted Dataset) 的抽象，它是分布在一组节点中的只读对象集合，这些集合是弹性的，如果数据集一部分丢失，则可以根据“血统”（即充许基于数据衍生过程）对它们进行重建。另 外在RDD计算时可以通过CheckPoint来实现容错，而CheckPoint有两种方式：CheckPoint Data，和Logging The Updates，用户可以采用哪种方式来实现容错。 最后，Spar k更加通用。不像Hadoop只提供了Map和Reduce两种操作，Spar

12、k提供的数据集操作类型有很多种，大致分为：Transfor mations和Actions两大类。Transfor mations包括Map、Filter、FlatMap、Sample、Gr oupByKey、ReduceByKey、Union、J oin、Cogroup、MapValues、Sor t和PartionBy等多种操作类型，同时还提供Count, Actions包括Collect、Reduce、Lookup和Save等操作。另外各个处理节点之间的通信模型不再像Hadoop只有Shuffle一种模式，用户可以命名、物化，、分区等。中间结果的18Briup Training 目前大数

13、据处理场景有以下几个类型： 1. 复杂的批量处理（Batch Data Processing），偏重点在于处理海量数据的能力， 至于处理速度可忍受，通常的时间可能是在数十分钟到数小时； 2. 基于历史数据的交互式分钟之间（Interactive Query），通常的时间在数十秒到数十 3. 基于实时数据流的数据处理（Streaming Data Processing），通常在数百毫秒到数秒之间 目前对以上三种场景需求都有比较成处理框架，第一种情况可以用Hadoop的MapReduce来进行批量海量数据处理，第二种情况可以Impala进行，对于第三中情况可以用Storm分布式处理框架处理实时流式

14、数据。交互式，各自一套维护成本比较高，而Spark的出现能够一站以上三者都是比较式平台满意以上需求。19Briup Training Spark是基于内存的迭代计算框架，适用于需要多次操作特定数据集的应用场合。需要反复操作的次数越多，所需的数据量越大，受益越大，数据量小但是计算密集度较大的场合，受益就相对较小 由于RDD的特性，Spark不适用那种异步细粒度更新状态的应用，例如web服或者是增量的web爬虫和索引。就是对于那种增量修改的应用模型不务的适合 数据量不是特别大，但是要求实时统计分析需求20Briup TrainingYahoo将Spark用在Audience Expansion中的

15、应用。Audience Expansion是广告中寻找目标用户的法：首先者提供一些了并且的样本客户，据此进行学习，寻找可能转化的用户，对他们定向。Yahoo采用的算法是logistic regression。同时由于有些SQL负载需要更高的服务质量，又加入了专门跑Spark的大内存集群，用于取代商业BI/OLAP工具，承担报表/图表和交互式/即席，同时与桌面BI工具对接。目前在Yahoo部署的Spark集群有112台节点，9.2TB内存。业务，最初使用Mahout或者写的MR来解决复杂的阿里搜索和学习，导致效率低而且代码不易维护。淘宝技术团队使用了Spark来解决多次迭学习算法、高计算复杂度的

16、算法等。将Spark运用于淘宝的推荐相关代的算法上,同时还利用Graphx解决了许多生产问题，包括以下计算场景：基于度分布的中枢节点发现、基于最大连通图的社区发现、基于三角形计数的关系衡量、基于随机游走的用户属性等。21杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：22Briup Training Spark运行

17、模式 Local本地模式(多用于测试) Standalone模式 Yarn集群模式 Mesos集群模式 Kubernetes集群模式 Spark环境搭建 Spark常用命令 spark-shell spark-submit spark-sql Spark基础案例23Briup Training 1.解压tar -xvf spark.2.2.1-bin-hadoop2.7.tar 2.mv spark.2.2.1-bin-hadoop2.7 /opt/software/ 3.修改环境变量： vi .bashrc export SPARK_HOME=/usr/spark/spark-2.2.1-b

18、in-hadoop2.7 exportPATH=$PATH:$SPARK_HOME/bin:$PATH:$SPARK_HOME/sbin 4.source .bashrc 5.(可选)修改下log4j的级别 $SPARK_HOME/conf/pertie 6.测试spark是否安装 spark-shell：24Briup Training 测试环境run-example SparkPi 10 测试spark-submit命令 $Spark_home/spark-submit-class org.apache.spark.examples.SparkPi examples/ja

19、rs/spark-examples_2.11-2.2.1.jar1025Briup Training 1.修改Spark配置文件 (路径为$SPARK_HOME/conf/下)#cp cpslaves.template和 spark-env.sh.template各一份spark-env.sh.template spark-env.shslaves.template slaves 2.vi slaves，此文件是指定子节点的主机,直接添加从节点主机名即可(前提是在/etc/hosts里面配置了).例如：mm0 mm1mm226Briup Training3.在spark-env.sh末端添加如

20、下几行：#必须配置：#主节点的IP地址export SPARK_MASTER_IP=00#主节点的端export SPARK_MASTER_PORT=7077#指定Spark用来混洗数据的本地路径export SPARK_LOCAL_DIRS=/data/spark/dirs,/home/briup/spark/dirs（一定要注意这个混洗数据的路径的权限）27Briup Training#可选配置：#Worker的WebUI端export SPARK_WORKER_WEBUI_PORT=8081#主节点的WEBUI端export SPARK_MASTER_WEBUI_P

21、ORT=8099#每个Worker使用的CPU核数,默认1个export SPARK_WORKER_CORES=2#每个Slave中启动几个Worker实例,默认1个export SPARK_WORKER_INSTANCES=2#每个Worker使用多大的内存,默认1g export SPARK_WORKER_MEMORY=2g #驱动器节点Driver使用的内存export SPARK_DRIVER_MEMORY=2g28Briup Training 4.启动start-all.sh命令或者是start-master.sh/start-slave.sh spark:/host:7077 5.

22、测试是否上输入jps页面，在集群的所有 6.打开localhost:8080查看集群29Briup Training 注意事项： 1.默认情况下，集群和Mesos提交的Application会使用所有集群中剩余的CPU Core。如果要限定每个提交的Spark Application的使用的CPU核的数目,可以采用下边的办法： 在sparkconf对象中设置spark.cores.max 核数。例如:val conf = new SparkConf() .setMaster(.).setAppName(.) .set( spark.cores.max , 10 ) 在群集上更改应用程序设置应该

23、使用spark.deploy.defaultCores。例如：在conf/ spark-env.sh： export SPARK_MASTER_OPTS=“-Dspark.deploy.defaultCores= 在spark2.2.1版本之后，需要在使用spark-submit或者spark-shell命令时，通过 total-executor-cores 2来进行设置。 2.必须确保executor_memory的值= worker_memory的值30Briup Training基于文件系统的单点恢复(Single-Node Recovery with Local File System

24、)export SPARK_DAEMON_JAVA_OPTS=-Dspark.deploy.recoveryMode=FILESYSTEM -Dspark.deploy.recoveryDirectory=/opt/software/spark- 1.6.1-bin-hadoop2.6/recovery基于zookeeper的Standby Masters(Standby Masters with ZooKeeper) export SPARK_DAEMON_JAVA_OPTS=-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zooke

25、eper.url=17:2181, 18:2181,19:2181 -Dspark.deploy.zookeeper.dir=/spark:用于指定Master的HA，依赖于zookeeper集群;31Briup Training 测试spark-submit命令 $Spark_home/spark-submit-master spark:/host:7077-class org.apache.spark.examples.SparkPi examples/jars/spark-examples_2.11-2.2.1.jar10

26、32Briup Training 1.只需要修改spark-env.sh环境配置即可： 2.添加如下配置：export HADOOP_CONF_DIR =/opt/hadoop/etc/hadoopexport SPARK_LOCAL_IP=你的IP 优化配置#将spark的类库jar包上传到hdfs上，省的每次都上传Spark1.*版本export SPARK_JAR=hdfs:/:9000/user/spark/share/lib/spark- assembly.jarSpark2.*版本spark.yarn.jars或者spark.yarn.archivehdfs:/:9000/spar

27、k_lib/jars/*hdfs:/:9000/spark_lib/jars33Briup Training 注意事项： 在yarn集群上部署spark,即添加HADOOP_CONF_DIR到spark-env.sh中, 不论要运行本地spark-shell还是yarn集群spark-shell,必须先启动yarn集群。34Briup Training 1.将hive-site.xml添加到$SPARK_HOME/conf/目录下； 2.集群运行模式下，需要将Hive的元数据的连接jar包配置到执行器节点； spark.executor.extraClassPath sql-connector

28、-java-5.1.22.jar 3.设置warehouse$HIVE_HOME/lib/my spark.sql.warehouse.dir hdfs:/computer1.cloud.briu :9000/user/hive/warehouse/ 4.测试 在本地测试只能通过 -driver-class-path 选项将Hive的元数据的连接jar包添加到Driver中。 spark-sql-driver-class-path $HIVE_HOME/lib/mysql-connector-java-5.1.22.jar35Briup Trainingspark-shellspark-sub

29、mit spark-sql36Briup Training 格式：./bin/spark-shell optionsoptions可选参数 可通过spark-shell-help查看eg:spark-shell -master local3 -name mySparkShell37Briup Training 格式：Usage: spark-submit options app argumentsUsage: spark-submit -kill submission ID -master spark:/. Usage: spark-submit -status submission ID -

30、master spark:/.Usage: spark-submit run-example options example-class exampleargsoptions可选参数 可通过spark-submit -help查看表示包含应用的jar包app arguments传给应用的选项38Briup Training spark-submit的常见标记-master 集群URL-deploy-mode 工作模式-class 应用程序的主类-name 应用的名称-jars 额外依赖的第jar包-files 需要分发到各节点的数据文件-executor-memory 执行器内存大小-driv

31、er-memory 驱动器内存大小-conf prop=value标记设置SparkConf配置选项-properties-file 指定一个包含键值对的属性文件. eg:spark-submit-master local -class com.briup.scala.First*.jar39Briup Training 格式：./bin/spark-sql options cli optionoptions可选参数 可通过spark-sql-help查看cli options可选参数可通过spark-sql-help查看eg:spark-sql-master local40Briup Tra

32、ining Scala行数统计val lines=sc.textFile(“ readme.md” ) lines.count()lines.first()思考：1. 哪来的sc?2. sc是什么?41Briup Training 驱动器程序(Driver Program) 包含应用的main函数，并定义了集群上的分布式数据集，并对分布式数据集应 用相关操作。 执行器(executor) sc(SparkContext)42Briup Training构建SparkContext对象 在一个的项目中，需要 pom文件中添加spark-core依赖org.apache.sparkspark-co

33、re_43Briup Training 代码编写结构如下：SparkConfval conf=newSparkConfval sc=newSparkContext(conf)SparkContextval rdd:RDDT=sc.方法(参数)RDDsc.stop()关闭SparkContext对象44Briup Training45Briup Training46杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-80

34、00：215311Brighten Your Way And Raise You Up.电邮：training主页：47Briup TrainingBriup Training Spark架构图49Briup Training50Briup Training Executor:执行器，Application运行在worker节点上的一个进程，负责启动线程池运行任务(Task),并且负责将数据存在内存或磁盘上。每个Application拥有executors。在Spark on Yarn 模式下， 其进程名称为的一组CoarseGrainedExecutorBackend， 类似于 Hadoop

35、 MapReduce 中的 YarnChild。一个 CoarseGrainedExecutorBackend 进程有且仅有一个executor 对象，它负责将 Task 包装成 taskRunner，并从线程池中抽取出一个空闲线程运行 Task。每个CoarseGrainedExecutorBackend 能并行运行 Task 的数量就取决于分配给它的CPU 的个数了；51Briup Training 弹性分布式数据集(RDD)Resillient Distributed Dataset，Spark 的基本计算单元，可以通过一系列算子进行操作（主要有 Transformation 和 Act

36、ion 操作,一组RDD形成执行的有向无环图RDD Graph。 有向无环图(DAG)DirectedSched Acycle graph，反应 RDD 之间的依赖关系； RDD DAG当RDD遇到Action算子，将之前的所有算子形成一个有向无环图。在Spark中转化为Job，提交到集群进行执行。一个App中可以包含多个Job。 有向无环图调度器(DAG Scheduler)根据 Job 构建基于 Stage 的 DAG，并提交 Stage给 TaskScheduler; 任务调度器(Task Scheduler):将Task分发给Executor执行。(将Taskset提交给worker(

37、集群)运行并回报结果); SparkEnv线程级别的上下文，运行时的重要组件的。52Briup Training53Briup Training54Briup Training Spark运行基本流程55Briup Training56Briup Training Spark on Standalone运行过程57Briup Training58Briup Training59Briup Training60Briup Training 在YARN中，每个Application实例都有一个ApplicationMaster进程，它是Application启动的第一个容器。它负责和Resourc

38、eManager打交道并请求之后告诉NodeManager为其启动Container。从获取次的含义讲YARN-Cluster和YARN-Client模式的区别其实就是ApplicationMaster进程的区别。 YARN-Cluster模式下，Driver运行在AM(Application Master)中，它负责向YARN申请，并监督作业的运行状况。当用户提交了作业之后，就可以关掉Client，作业会继续在YARN上运行，因而YARN-Cluster模式不适合运行交互类型的作业； YARN-Client模式下，Application Master仅仅向YARN请求Executor，Cli

39、ent会和请求的Container通信来调度他们工作，也就是说Client不能离开。61杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：62Briup Training 1.抽象模型RDD 2.RDD操作构建RDDRDD方法闭包基于分区进行操作共享变量普通RDD常见方法键值对RDD特有的方法 3.PageRank案

40、例/ 4.Spark配置与调优评分数据分析63杰普软件科技公司总部：上海市闸北区万荣路1188弄龙软软件园区A栋206室：（021）56778147：200436昆山实训基地：昆山市巴城学院路828号昆山浦东软件园北楼4-5层：（0512）50190290-8000：215311Brighten Your Way And Raise You Up.电邮：training主页：64Briup Training RDD：弹性分布式数据集（Resilient Distributed Dataset） RDD是Spark中最基本的数据抽象，它代表一个不可变(只读)、可分区、元 素可并行计算的,完全弹性

41、的集合。 RDD是有属性的数据集。属性当前数据集的状态，数据集是由数据的分区（partition）组成, 这些分区运行在集群中的不同节点上 5个属性：1. partitions 数据分区列表pute 对分区数据计算函数3.dependencies 依赖列表4.partitioner 分区方式5.preferredLocations 优选位置RDD具有数据流模型的特点：自动容错、位置感知性调度和可伸缩性。RDD用户在执行多个时显式地将工作集缓存在内存中，后续的能够重用工作集，这极大地提升了速度。65Briup Training RDD之所以称为“ 弹性” 的特点 :1. 基于Lineage的高效

42、容错（第n个节点出错，会从第n-1个节点恢复，血 统/缘容错）;2. Task如果失败会自动进行特定次数的重试（默认4次）;3. Stage如果失败会自动进行特定次数的重试（可以只运行计算失败的阶 段），只计算失败的数据分片;4. 数据调度弹性：DAG TASK 和5. checkpoint检查点容错;6. 自动的进行内存和磁盘数据管理无关;的切换.66Briup Training 对开发者而言: 1.RDD可以看作是Spark的一个对象，它本身运行于内存中。 如读文件是一个RDD，对文件计算是一个RDD，结果集也是一个RDD 不同的分片、 数据之间的依赖 、key-value类型的map数据

43、都可以看做RDD。 2.RDD不需要物化。 RDD含有如何从其他RDD衍生（即计算）出本RDD的相关信息（即Lineage），据此可以从物理的数据计算出相应的RDD分区 。 RDD作为数据结构，本质上是一个只读的分区集合。 一个RDD可以包含多个分区，每个分区就是一个dataset片段。 RDD可以相互依赖。 如果RDD的每个分区最多只能被一个Child RDD的一个分区使用，则称之为narrow dependency； 若多个Child RDD分区都可以依赖，则称之为wide dependency。 不同的操作依据其特性，可能会产生不同的依赖。例如map操作会产生narrow depende

44、ncy，而join操作一般则产生wide dependency。67Briup Training68Briup Training69Briup Training70Briup Training Partition结构/* An identifier for a partition in an RDD.*/trait Partition extends Serializable /* Get the partitions index within its parent RDD*/def index: Int/ A better default implementation of HashCode

45、 override def hashCode(): Int = indexoverride def equals(other: Any): Boolean = super.equals(other)71Briup Training72Briup Training 2）compute 对分区数据计算函数。Spark中RDD的计算是以分区为的，每个RDD都会实现compute函数以达到这个目的。compute函数会对迭代器 进行复合，不需要保存每次计算的结果。 compute(Partition,TaskContext)73Briup Training 3）dependencies 依赖列表： R

46、DD之间的依赖关系。RDD的每次转换都会生成一个新的RDD，所以RDD之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时， Spark可以通过这个依赖关系重新计算丢失的分区数据，而不是对RDD的所有分区进行重新计算。74Briup Training 1.RDD操作-转化操作与行动操作 2.RDD缓存机制 3.DAG 有向无环图(任务执行机制) 4.lineage/谱系图/血缘系统(容错机制) 5.依赖关系 窄依赖(NarrowDependency):父RDD一个数据分区只被RDD的一个数据分 区所使用。filter map等方法 OneToOneDependency RangeD

47、ependency 宽(Shuffle)依赖(ShuffleDependency):父RDD一个数据分区被子RDD的多个数据分区所使用。groupByKey sortByKey等方法75Briup Training76Briup Training77Briup Training 4）partitioner 分区方式： 1.Spark程序通过RDD分区方式来减少通信开销 2.只有对于于key-value的RDD，才会有Partitioner 3.非key-value的RDD的Parititioner的值是None。 4.Partitioner函数不但决定了RDD本身的分区数量，也决定了paren

48、t RDD Shuffle输出时的分区数量。 Partitioner源码实现abstract class Partitioner extends Serializable def numPartitions: Intdef getPartition(key: Any): Int78Briup Training 1.Spark内置支持的两种分区方式，一个是基于哈希的HashPartitioner，另外 一个是基于范围的RangePartitioner。 2.查看当前RDD分区方式:rdd.partitioner 3.自定义分区方式:extends Partitioner重写两个抽象方法即可 4.

49、设置好RDD分区的操作: cogroup()、groupWith()、join()、leftOuterJoin()、rightOuterJoin()、groupByKey()、reduceByKey()、combineByKey()、partitionBy(),sortByKey() 以及mapValues()(如果父RDD有分区方式的话) flatMapValues()(如果父RDD有分 区方式的话),以及filter()(如果父RDD有分区方式的话)等，其他操作生成结果都存在特定分区方式。 如果RDD在调用以上操作之前已经具有的分区方式，则以上操作都能够从分区中获益。(此时以上方法就进行数

50、据混洗，减少了数据混洗的开销)79Briup Training 5.预定义自定义分区(一般预定义分区与缓存方法连用) pairRDD.partitionBy(args:Partitioner).cache/persist() 适用场景:迭代计算中反复使用同一RDD调用带有shuffle操作的方法时，就 可以在进入迭代计算前，先对该RDD进行预定义分区，并记住一定要调用缓存方法，将该RDD缓存起来，减少数据混洗的开销。 举例:val newRDD=rdd.partitionBy().cache(); (1 to 50).foreach(x=newRDD.groupByKey.collect /1

51、 次Shuffle) 对于二元操作，输出数据的分区方式取决于父RDD的分区方式。默认情况下， 结果采用哈希分区，分区的数量和并行度一样。 如果其中一个父RDD已经设置过分区方式，那么结果就会采用那种分区方式。 如果两个父RDD都设置过分区方式，结果RDD采用第一个父RDD的分区方式。80Briup Training 5）preferredLocations 优选位置：每个Partition的优先位置的列表（preferred location）。对于一个HDFS文件来说，这个列表保存的就是每个Partition所在的块位置。 1.数据文件所在块的位置信息。 2.按照 移动数据不如移动计算 原则,Spark在进行任务调度的时候，会尽