Spark集群.pdf

《Spark集群.pdf》由会员分享，可在线阅读，更多相关《Spark集群.pdf（21页珍藏版）》请在得力文库 - 分享文档赚钱的网站上搜索。

1、 1.Spark 集群+流水线+RDD 1.1.独立集群方式 1.1.1.独立集群方式 集群搭建分了三种方式：1）独立集群方式 StandAlone 2）和 Yarn 进行整合，由 Yarn 来完成资源的管理和调度，和 hadoop 无缝集成 3）和 Apache Mesos 进行整合，由 Mesos 进行管理和调度，轻松 3 万个节点 1.1.2.修改每台的配置文件 vi conf/spark-env.sh 设置：SPARK_LOCAL_IP=192.168.1.106#注意每台配置的是本机 IP 地址 还可以配置 CPU 和内存，根据实际情况进行配置，不配置时，默认读取当前服务器的实际情况

2、。SPARK_WORKER_CORES、SPARK_WORKER_MEMORY。1.1.3.修改每台的 hosts 文件 vi/etc/hosts 192.168.1.106 spark1 192.168.1.108 spark2 192.168.1.105 spark3 192.168.1.107 spark4 192.168.1.103 hadoophdfs 配置集群中其它主机的 IP 地址和机器名，每台的服务器的 hosts 都要进行修改 1.1.4.启动 Master 任意选择一台服务器作为 Master，其它就都作为 worker ssh 192.168.1.106 cd/usr/l

3、ocal/src/spark/spark-1.5.2-bin-hadoop2.6 sbin/start-master.sh h 192.168.1.106#启动 master sbin/stop-master.sh#停止 master 注意：可以是 IP 地址也可以是主机名 启动后显示日志文件路径。要学会看错误日志。这样当出现错误，可以知道问题出在哪里了。注意：1）akka 访问地址 work 连接的地址 2）WebUI 访问地址，可以通过浏览器查看集群状态（单机是没有这个管理界面的）1.1.5.每台服务器防火墙打开端口 akka 访问端口：7077 和 WebUI 控制台 Web 服务端口

4、8080 service iptables stop#简单测试时，直接关掉防火墙 web 管理界面 http:/192.168.1.106:8080/1.1.6.启动 Worker 在各个 work 上启动各自的 worker sbin/start-slave.sh spark:/192.168.1.106:7077#指向 master 的访问地址#sbin/start-slave.sh spark:/192.168.1.106:7077 h spark1 指定 IP 地址或者主机名访问，连接 HDFS 按主机名会刚好些，IP 可能出错。sbin/stop-slave.sh#停止服务 有时第一

5、次连接时不成功，重新连接次就好了。同样也会有日志输出，刚开始你不熟悉时，最好看一下日志，看下启动时有没有错误，如果你配置有错，或者连接地址有错，或者防火墙没开，它都有可能连接不到 master。所以必须先看下日志，养成一个好习惯。tail f 日志文件路径 可以看到它也会启动一个 Web 管理界面。看到“successful regiestered”搞定。刷新 Master，可以看到有 worker 就管理起来了。State 状态 ALIVE 就表示活着，将来集群很多 worker，master 只会把任务分配给 ALIVE 活着的。如果 DEAN，就不会被分配。Cores 使用 CPU 的个

6、数 Memory 是使用内存的大小 1.1.7.修改配置自动刷新 sbin/stop-slave.sh#关闭服务 增加 CPU 的核数 4，增加内存到 2G sbin/start-slave.sh#启动服务 刷新控制台，可以看到 CPU 和内存都加大了。可以看到旧的已死，新的已启动。启动较慢，稍等几秒。1.1.8.客户端如何连接到集群中进行计算 bin/spark-shell-master=spark:/192.168.1.106:7077#master 的 IP 地址 可以看到管理界面上出现一个 RunningApplication，正在执行的应用。SparkShell就是一个特殊的应用程序

7、，它可在集群环境中运行。1.1.9.运行任务 val rdd=sc.parallelize(List(1,2,3,4,5,6),2)rdd.collect 可以看到进行了两个分区，上面刚好有两个 worker 是存活的。1.1.10.查看执行结果，必须点“Spark shell”进入 执行的 collect 操作 Job 一次完整的计算称为一个 job Submitted 提交任务的时间 Duration 耗时：这个 2s 是很慢的，因为我们在虚拟机中。Stages 执行的阶段，现在的任务比较简单，执行很快，所以就一个阶段，复杂任务会有多个。Tasks 任务：成功执行的任务/总共的任务数。1.

8、1.11.常见错误 failed to bind 如果配置了 SPARK_LOCAL_IP,但是并没有在 slaves 上修改为自己的 IP,则会报错:ERROR netty.NettyTransport:failed to bind to/192.168.6.52:0 .BindException:Failed to bind to:/192.168.1.105:0:Service sparkWorker failed 1.2.*流水线 pipeline 1.2.1.流水线 pipeline Spark 为何比 Hadoop 快？1.2.2.懒惰式命令 一开始学习时容易产生困扰的地方 执行

9、rdd.map，或者 println 都没有发现有动作，刷新 job 也看不到，实际上就没有 立即执行或者说它的行为是懒惰式的。因为它并没有构成一个完整的任务。直到 spark 执行另一类方法时，才会触发计算。例如：rdd.count 或者 rdd.collect。这些方法才会触发计算，才把它看做是一次完整的计算。可以看出 map 不是一次完整的计算。1.2.3.Spark 中把方法分为了两类 一类 transformations 不会触发 job 的提交（没有立即计算），例如：map、filter、flatMap、reduceByKey 等。一类 actions 会触发 job 的提交，例如

10、：collect、count 等。Spark 在执行真正任务之前，都会做一些优化：会计算 rdd 之间的依赖关系，然后根据依赖关系划分阶段，完成优化。1.2.4.*集群实例 1）创建一个 letter.txt 1,a 2,b 3,c 4,d 5,e 3,f 2,g 1,h 注意它的环境，现在是在集群环境中。假如 letter.txt 在 spartk2 的服务器上，那spark3 的服务器呢，肯定没有这个文件就会报错。如何解决呢？这是分布式计算时，需要注意的问题。要解决这个问题 sc.textFile()所要读取的文件必须在每个服务器上都存在。这也是初学者容易犯的错误。2）准备：创建工具类，工

11、具类会触发 job import org.apache.spark.rdd.RDD import scala.reflect.ClassTag object su def debugT:ClassTag(rdd:RDDT)=rdd.mapPartitionsWithIndex(i:Int,iter:IteratorT)=val m=scala.collection.mutable.MapInt,ListT()var list=ListT()while(iter.hasNext)list=list:+iter.next m(i)=list m.iterator ).collect().forea

12、ch(x:Tuple2Int,ListT)=val i=x._1 println(spartition:$i)x._2.foreach println )3）sc.textFile(/root/letter.txt)它也是不会立即执行，等 job 触发时才执行 scala sc.textFile(/root/letter.txt)res1:org.apache.spark.rdd.RDDString=MapPartitionsRDD1 at textFile at:16 scala su.debug(res1)partition:0 1,a 2,b 3,c 4,d 5,e partition:

13、1 3,f 2,g 1,h 说明它把这个文件，前 5 行分到一个分区，后 3 行分到一个分区。4）res1.map(x=val a=x.split(,);(a(0),a(1)将原始的字符串转换为 tuple 类型 partition:0 (1,a)(2,b)(3,c)(4,d)(5,e)partition:1(3,f)(2,g)(1,h)5）res7.map(x=(x._1,x._2+_s)partition:0 (1,a_s)(2,b_s)(3,c_s)(4,d_s)(5,e_s)partition:1(3,f_s)(2,g_s)(1,h_s)1.2.5.*依赖关系 前面的步骤都没有触发jo

14、b，它们这几步都在创建依赖关系，每一步都在生成新的RDD。那如何看这个关系呢？找到最后一个 RDD，它的 toDebugString scala res12.toDebugString res16:String=(2)MapPartitionsRDD6 at map at:22|MapPartitionsRDD4 at map at:20|MapPartitionsRDD1 at textFile at:16|/root/letter.txt HadoopRDD0 at textFile at:16 最后生成的 RDD 叫做 HadoopRDD 对象，它既可以读 Hadoop 文件也可以读普通

15、文件。这是 spark 为我们优化的第一步，它就是 transformations 类型，每一步都会生成新的 RDD 对象，每一个 RDD 不是独立的，它必须依赖上一个 RDD。或者称为父 RDD 对象。它们之间都有父子关系。spark 就会应用一种叫流水线的技术，它特别的节省内存，并且也让性能得到提升。scala res1.partitions.length res19:Int=2 scala res7.partitions.length res20:Int=2 scala res12.partitions.length res21:Int=2 每个 RDD 有两个分区。分区个数由父 RDD

16、 决定。上面的过程只是绘制了一个美好的蓝图，并未执行，那什么时候会真正执行运算呢？那就是执行一个 actions 类型命令时才会运算。res12.collect 触发一次完整的计算，根据分区个数生成任务，根据 2 个分区生成了两个任务（Task）。spark 就把它封装成任务对象，底层通过 akka 框架发给每个 worker。Task 对象中就包括了所有的 RDD 信息。Task 最初在 Driver 上（SparkShell）。真正执行不是在 SparkShell，而是在 worker 上执行。怎么发？就是 akka 框架，先把它序列化为二进制的，再把它发给 worker，worker 收

17、到二进制的，反序列化，就是 Task。RDD 都在 Task 里面，它们之间的关系就知道了。具体计算是怎么计算的呢？它采用的是倒推的方式进行最后的计算。它先看最后一个RDD，res12 去执行时，遇到变量 x，它就去读 res7，res7 遇到变量 x，就去读取 res1，res1 遇到 textFile 就按事先分区好的去读取前 5 行。（文件总共 8 行，前 5 行分给第一个分区，后 3 行分给第二个分区。）每次读一行，因为 res7 的 x，每次只要一个。1.2.6.总结优缺点 不会一次读很大量的数据，不会造成内存溢出 中间结果流动方式无需暂存，内存消耗小 命令批量提交，性能高 Spar

18、k 首先采用了迭代思想，会不会用很大的内存？不会，数据是一条一条处理的。它边读边进行运算。它就不需要中间暂存结果。传统的 Hadoop 为什么慢，它每算一步，中间的结果都要暂存在 HDFS 上。为什么说 spark 效率好呢？就上面的例子，中间的结果用不用暂存？中间的处理是以流动的方式，一个 RDD 传到下一个 RDD，就不需要暂存了。因此它能够充分的利用内存。这是它效果高的因素之一。这种方式就称为流水线（pipeline）。从这也看出来，spark 为何要把一些命令设置为 transformation 方式，先不执行，最后批量一次执行。你要立即执行，就得保存中间结果，供下一次执行使用。缺点：

19、每个分区的数据在一个 worker 上运行，万一有一个 worker 出现了故障，会导致出错的 worker 还需要重新计算，因为没有中间结果，一旦 worker 运算失败，spark 需要找一个正常的 worker 来重新计算，而且从头算，这样就拉长了总的计算时间。当然这个缺点可以配合一些缓存，例如一些中间结果非常重要，可以再加上缓存。以后出现故障不用重新算，从这些“保存点”上重新计算。有些类似有些的中间存盘。后面会详细讲到，加缓存也是 spark 很重要的一块，也是比 hadoop 性能高的原因之一。它的缓存是在内存，比缓存到 HDFS 上快的多。1.3.和 HDFS 结合 1.3.1.和

20、 HDFS 结合 上面的例子有个问题，就是需要处理的问题，必须保证每个 worker 中都有一个，那这种方式及其不方便，而且处理之前难以保证每个服务器都有。如何解决？可以使用 HDFS 分布式文件系统。1.3.2.准备：伪分布 Hadoop 单机 服务访问地址：http:/192.168.1.103:50070/检查 HDFS 环境是否正常 jps#查看 jvm 进程状态 18406 ResourceManager 18253 SecondaryNameNode 18856 Jps 18085 DataNode 18508 NodeManager 16364 NameNode 必须有上面 6

21、个进程才正常 cd/usr/local/src/hadoop/hadoop-2.7.1 bin/hdfs dfs mkdir/test#创建 test 目录，注意必须前面加斜杠/bin/hdfs dfs ls/#浏览目录 上传 letter.txt 到/usr/local/src/hadoop/hadoop-2.7.1 目录 步骤：1、将测试文件放入到 HDFS 系统中 bin/hdfs dfs put letter.txt/test#letter.txt 放入 HDFS 的 test 目录下。2、在 master 的 spark 配置文件中配置（可以不配置）Spark 和 Hadoop 在一

22、台机器上。HADOOP_CONF_DIR，配置完 spark 就知道了 hadoop 的 namenode 的访问地址。就无需写代码连接 hadoop，它内部自动连接。sparkMaster 会以广播的形式将连接信息广播到其他所有的 worker 上。HADOOP_CONF_DIR=/root/hadoop-2.7.1/etc/hadoop 注意不是配置 hadoop 目录，是配置 hadoop 配置文件所在目录 etc/hadoop。就是我们通常修改 core-site.xml 和 hdfs-site.xml 文件所在路径。3、重启 master、slave 上面不配置，也无需此步。4、读取

23、 HDFS 文件 必须在每台服务器上的/etc/hosts 文件加上 192.168.1.103 hadoophdfs sc.textFile(hdfs:/hadoophdfs:9000/test/letter.txt)#此时未读 res0.collect#执行文件读取，这样就检查了，上面的命令是否正确 LocalityLevel 如果 worker 和 datanode 不在一台服务器，就是 ANY；如果在一起就是 PROCESS_LOCAL/NODE_LOCAL。在一起是最好的，无需网络传输，访问最快。1.3.3.扩展：worker 和 datanode 什么样的关系？worker 和 d

24、atanode 在一台服务器上是最后的，这样无需网络传输，访问最快。数据尽量本地化。本地化程度由高到低：序号 名称 说明 1 PROCESS_LOCAL 表示 HDFS 数据和 worker 是在同一个进程 2 NODE_LOCAL 表示 HDFS 数据和 worker 是在同一台服务器 3 NO_PREF 表示 spark 不能确定数据的位置 4 RACK_LOCAL 表示 HDFS 的数据和 worker 在同一个机架内 5 ANY 表示 HDFS 的数据和 worker 不在同一个机架内 NODE_LOCAL 一般可以保证，通过配置设定 先检查 worker 是否启动 即使 work 和

25、 datanode 在一台服务器上，但按 ip 地址启动也不能生效，还是 ANY。这应该和 hadoop 相关，一般 hadoop 中使用主机名去访问 namenode，没有用 ip 地址去访问。sbin/stop-slave.sh sbin/start-slave.sh spark:/192.168.35.4:7077 h spark4 感觉比刚才快些。可以看出有两个 worker，一个本地一个远程，spark 都不把任务交给远程的 worker直接本地执行了。它发现 spark4 数据和 datanode 在同一台服务器上。它就优先使用本地的。这样无需跨网络，性能就高。这是 spark 中

26、一个相当优化的措施。这个不是 spark 自己实现的，它还是调用 hadoop 的 api 来实现判断是否是同一台服务器。总结：在启动 slave 时多加-h 参数指定主机名，就可以实现 NODE_LOCAL 设置。1.3.4.虚拟机重启后执行步骤 1、检查配置每个的主机名 hosts 2、启动 HDFS 服务 服务访问地址：http:/192.168.1.103:50070/3、启动 SparkMaster 4、启动 SparkSlave 5、启动 SparkShell 1.4.RDD 的常用方法 RDD 有两类 1）RDD 是处理非键值对的 http:/spark.apache.org/d

27、ocs/latest/api/scala/index.html#org.apache.spark.rdd.RDD 2）PairRDDFunctions 处理键值对的 http:/spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.rdd.PairRDDFunctions 1.4.1.+并集 等价于 union 将两个 rdd 做并集 例子：val rdd1=sc.parallelize(List(1,2,3,4,5)rdd1.partitions.length#分区是 3，两个 worker 的 core 总数

28、val rdd2=sc.parallelize(List(5,6,7,8,9,10)rdd2.partitions.length val rdd3=rdd1+rdd2#rdd1.union(rdd2)rdd3.collect 结果：res4:ArrayInt=Array(1,2,3,4,5,6,7,8,9,10)它使每个分区不变，就是将分区合起来使用。可以看出，它就没有动每个分区，就是把分区合起来。大家思考下如果动了什么结果？分区如果你的数据发生变了，就可能会发生网络传输了。1.4.2.collect 收集 等价于 union 将两个 rdd 做并集 1.4.3.take 获取元素 take(

29、n)从分区中获取元素 scala val rdd=sc.makeRDD(List(1,2,3,4,5)rdd:org.apache.spark.rdd.RDDInt=ParallelCollectionRDD6 at makeRDD at:21 scala rdd.take(2)res12:ArrayInt=Array(1,2)1.4.4.takeOrdered(n)升序 从分区中获取头 n 个元素进行排序。先排序后获取。scala val list=sc.parallelize(List(4,5,6,3,2,11)list:org.apache.spark.rdd.RDDInt=Parall

30、elCollectionRDD9 at parallelize at:21 scala list.take(3)res16:ArrayInt=Array(4,5,6)scala list.takeOrdered(3)res17:ArrayInt=Array(2,3,4)takeOrdered 的内部实现是先分 3 个区，每个区里拿出三个最小的，没有那么多就都拿，再把这 9 个值比较下，在这 9 个里再取最小的。1.4.5.top(n)降序 它的过程和上面相反，先分三个分区，三个分区中找出最大的，然后再比较，最后得到最大的。scala val list=sc.parallelize(List(4

31、,5,6,3,2,11)list:org.apache.spark.rdd.RDDInt=ParallelCollectionRDD9 at parallelize at:21 scala list.top(3)res18:ArrayInt=Array(11,6,5)1.4.6.*cache 缓存 将 RDD 的内容缓存，这样就无需继续往父 RDD 找了，直接去缓存中获取值。scala var rdd1=sc.textFile(README.md)rdd1:org.apache.spark.rdd.RDDString=MapPartitionsRDD1 at textFile at:21 sc

32、ala val rdd2=rdd1.filter(x=println(Thread.currentThread);x.contains(a)rdd2:org.apache.spark.rdd.RDDString=MapPartitionsRDD5 at filter at:23 scala rdd2.cache scala rdd2.count res0:Long=18 scala rdd2.count res0:Long=18 例如：观察运行结果，发现第一次执行 count 时，控制台输出了相关线程信息，第二次执行 count 时，不会再次出现线程信息，其实已经从缓存中进行读取。怎么查看 c

33、athe 呢？通过管理界面。它存储可能分为两部分，一部分存在内存中，如果内存不够，会强制保存到磁盘。可以看到执行完 cathe，类型就是 MapPartitionsRDD。scala val rdd2=rdd1.filter(x=println(Thread.currentThread);x.contains(a)rdd2:org.apache.spark.rdd.RDDString=MapPartitionsRDD5 at filter at:23 缓存也是分区的 Cached Partitions。cache 很重要，因为它也是 spark 比 hadoop 快的一个原因之一。它把经常需要

34、使用的内容放入缓存中，这样无需再次计算，就快了。注意：给那些需要大量重复的 RDD 使用缓存。1.4.7.*persist 缓存 比 cache 多些选项，支持压缩，支持外部磁盘等。如果缓存的内容非常大怎么办？可以使用很多的策略，一个就是压缩。比如有 1 万个对象，那内存中放不下，怎么办？就对这 1 万个对象进行压缩。压缩完就 500M 了，那它就可以放到内存中，这样也有助于性能的提升。这个压缩不要小看它，spark 要充分的利用内存，如果内容很多，内存放不下，那一部分就需要放到磁盘上，那完了，效率立刻下来了。所以有时候即使压缩下，也比你放在硬盘上强。但压缩也必然需要解压缩，但它占用的是 CP

35、U 的资源。牺牲了 CPU，节省了内存。如果还是大大呢，压缩也不行了，可以多构造几个 partition 分区，它运行完一个分区，才运行另一个分区。注意先要导入包 StorageLevel，才能调用这几个常量 scala import org.apache.spark.storage.StorageLevel import org.apache.spark.storage.StorageLevel scala val rdd=sc.textFile(hdfs:/hadoophdfs:9000/test/letter.txt)rdd:org.apache.spark.rdd.RDDString=

36、MapPartitionsRDD1 at textFile at:22 scala rdd.persist(StorageLevel.MEMORY_ONLY_SER)res2:rdd.type=MapPartitionsRDD1 at textFile at:22 scala rdd.collect res3:ArrayString=Array(1,a,2,b,3,c,4,d,5,e,3,f,2,g,1,h)可以看到明显小多了，456B 变成 56B，尤其对文本压缩率相当高。1.4.8.扩展：spark 中对内存如何划分？多少作为缓存使用，多少作为计算使用？这个在 spark 中也有讲究。sp

37、ark 中有个configuration 对象，它提供了很多配置 spark.storage.memoryFraction 内存因子，缓存占内存多大。默认 0.6。例如：worker 1g 内存，60%是作为内存的，也就是 600m 作为缓存，剩下。这样可以事先计算好，让它别超过缓存。配置文件 spark-defaults.conf.template 文件中配置。可以自己加上，注意之间用空格区分，spark 要重新启动才生效。还需要复制配置文件为 spark-defaults.conf，否则也不生效。cp spark-defaults.conf.template spark-defaults.

38、conf 1.4.9.cartesian 笛卡尔积 男 male 女 female 发衣服，每人都有一件马甲 jacket 和裤子 trousers val rdd1=sc.parallelize(List(male,1),(female,1)val rdd2=sc.parallelize(List(jacket,1),(trousers,1)rdd1.cartesian(rdd2)res12.collect res14:Array(String,Int),(String,Int)=Array(male,1),(jacket,1),(male,1),(trousers,1),(female,1

39、),(jacket,1),(female,1),(trousers,1)1.4.10.coalesce(n,true/false)扩大或缩小分区 如果内存不够，可以重新分区，如果原来 3 个，可以重新分成 6 个。也可以减少分区，例如：运算过程中发现用不了那么多分区，也可以减少。可以从少变多，也可以从多变少。scala val rdd=sc.parallelize(List(1,2,3,4,5,6),2)rdd:org.apache.spark.rdd.RDDInt=ParallelCollectionRDD18 at parallelize at:22 scala rdd.partition

40、s.length res25:Int=2 scala rdd.coalesce(5)res27:org.apache.spark.rdd.RDDInt=CoalescedRDD19 at coalesce at:25 scala res27.partitions.length res28:Int=2 scala rdd.coalesce(4,true)res30:org.apache.spark.rdd.RDDInt=MapPartitionsRDD23 at coalesce at:25 scala res30.partitions.length res31:Int=4 由少到多要多加一个参

41、数 shuffle 为 true，默认是 false。由多到少，不需要。val rdd=sc.parallelize(List(1,2,3,4,5,6),4)rdd:org.apache.spark.rdd.RDDInt=ParallelCollectionRDD24 at parallelize at:22 scala rdd.partitions.length res32:Int=4 scala rdd.coalesce(2)res34:org.apache.spark.rdd.RDDInt=CoalescedRDD26 at coalesce at:25 scala res34.part

42、itions.length res35:Int=2 1.4.11.repartition(n)分区扩大=它等价于于 coalesce(n,true)由下面源码就可以看到，实际它内部就调用的 coalesce。1.4.12.count 计算 RDD 中对象个数 scala val rdd=sc.parallelize(List(1,2,3,4,5,6),3)rdd:org.apache.spark.rdd.RDDInt=ParallelCollectionRDD27 at parallelize at:22 scala rdd.count res36:Long=6 1.4.13.countApp

43、rox 计算一个近似值 当计算量非常大，不需要精确结果时，求一个近似值的情况下使用，设置超时时间毫秒数，超时时间越长，计算的越精确。1.4.14.intersection 交集 求两个 RDD 共有的元素 1.4.15.mapPartitionsWithIndex 分别遍历分区做不同的处理 分区 1 加 a，分区 2 加 b。1.4.16.saveAsTextFile 保存分区内容到磁盘 注意 aaa 是目录不是文件，它会自动起文件名 part-nnnn 保存本地文件有时候不成功 一般也都保存到 HDFS 上，有两个分区，就有两个文件。1.4.17.sortBy 排序 升序 降序 1.4.18.zip 拉链操作 两个集合元素个数得相同，然后合并成 tuple 类型