《大数据技术与应用基础-第8章开源集群计算环境spark》由会员分享，可在线阅读，更多相关《大数据技术与应用基础-第8章开源集群计算环境spark（28页珍藏版）》请在金锄头文库上搜索。
1、P123第8章开源集群计算环境Spark大数据技术与应用基础21世纪高等院校“云计算和大数据”人才培养规划教材第8章 开源集群计算环境SparkP123第8章开源集群计算环境Spark 能力CAPACITY要求了解如何根据实际需求选择合适的算 子操作数据。理解理解Spark的工作机制。了解使用Spark提供的算子对数据进 行转化。熟悉对源数据结构格式进行分析。P123第8章开源集群计算环境Spark 内容CONTENTS导航Spark接口实战Spark简介Spark编程之RDDSpark实战案例之统计1000万人口的平均年龄Spark MLlib实战（聚类实战）P123第8章开源集群计算环境S
2、park一、Spark简介简介Spark是UC Berkeley AMP lab所开源的类Hadoop MapReduce的 通用并行框架，Spark拥有Hadoop MapReduce的优点；但不同 于MapReduce的是Spark工作中间输出结果可以保存在内存中， 而不再需要读写HDFS。因此Spark能更好地适用于数据挖掘与机 器学习等需要迭代的MapReduce的算法。Spark可以在Hadoop文 件系统中并行运行。通过名为Mesos的第三方集群框架可以支持 此行为。P123第8章开源集群计算环境Spark 内容CONTENTS导航Spark简介Spark编程之RDDSpark实战
3、案例之统计1000万人口的平均年龄Spark MLlib实战（聚类实战）Spark接口实战P123第8章开源集群计算环境Spark二、Spark接口实战环境要求IDEA使用和打包（1）启动IDEA“Welcome to IntelliJ IDEA”“Create New Project”“Scala”“Non-SBT”创建一个 名为text的project（注意这里选择自己安 装的JDK和scala编译器） “Finish”。 （2）增加开发包：依次选择 “File”“Project Structure”“Libraries”“+”“java”选 择/opt /spark/spark-1.5.
4、2/lib/spark- assembly-1.5.2-hadoop2.6.0.jarUbuntu 16.04 server64 JDK 1.8.0 Hadoop 2.6.0 Spark 1.5.2 Scala 2.11.7 集成开发环境：IDEA 2016- IC（需安装scala插件）P123第8章开源集群计算环境Spark二、Spark接口实战IDEA使用和打包object WordCount def main(args: ArrayString) if (args.length = 0) System.err.println(“Usage: WordCount “)System.exi
5、t(1) val conf = new SparkConf().setAppName(“WordCount“)val sc = new SparkContext(conf)sc.textFile(args(0).flatMap(_.split(“).map(x= (x,1).reduceByKey(_+ _).take(10).foreach(println)sc.stop()（3）编写代码。在src目录下创建1 个名为text的package，并增加一个 Scala class类型为Object。实现单词计数的WordCount代码 ，如下所示： package text /* Create
6、d by hadoop on 9/1/16.*/ import org.apache.spark.SparkContext, SparkConf import org.apache.spark.SparkContext._P123第8章开源集群计算环境Spark二、Spark接口实战IDEA使用和打包单击“OK”按钮后，选择“Build”“Build Artifacts”“text”“build”进行打包， 经过编译后，程序包放置在 ideaProjects/text/out/artifacts/text目 录下，文件名为text.jar。 （5）运行。把spark自带文件 README.md
7、上传到hdfs，命令如 下。 hadoop fs put /opt/spark/README.md /user/hadoop（4）生产程序包。生成程序 包之前要先建立一个artifacts ，依次选择“File”“Project Structure”“Artifacts”“+”“ Jars”“From moudles with dependencies”，然后随便选 一个Class作为主Class；单击 “OK”按钮后，对artifacts进行 配置，修改Name为text，删除 Output Layout中text.jar中的几 个依赖包，只剩text项目本身 ；P123第8章开源集群计算环
8、境Spark二、Spark接口实战IDEA使用和打包运行完成后在Web监控界面可以看到结果信息，如下图P123第8章开源集群计算环境Spark 内容CONTENTS导航Spark接口实战Spark简介Spark实战案例之统计1000万人口的平均年龄Spark MLlib实战（聚类实战）Spark编程之RDDP123第8章开源集群计算环境Spark三、Spark编程之RDDRDDRDD是Spark中的抽象数据结构类型，从编程的角度来看， RDD可以简单看成是一个数组。和普通数组的区别是， RDD中的数据是分区存储的，这样不同分区的数据就可以 分布在不同的机器上，同时可以被并行处理。 因此，Spa
9、rk应用程序所做的无非是把需要处理的数据转换 为RDD，然后对RDD进行一系列的变换和操作从而得到结 果。下面，将介绍Spark RDD中与Map和Reduce相关的 API。创建RDD RDD可以从普通数组创建出来，也可以从文件系统或者 HDFS中的文件创建出来。 例1：从普通数组创建RDD，里面包含了1到9这9个数字，它 们分别在3个分区中。 scala val a = sc.parallelize(1 to 9, 3) a: org.apache.spark.rdd.RDDInt = ParallelCollectionRDD1 at parallelize at :12P123第8章开
10、源集群计算环境Spark三、Spark编程之RDDRDD中与Map和Reduce相关的API1map map是对RDD中的每个元素都执行一个指定的函数来产生 一个新的RDD。任何原RDD中的元素在新RDD中都有且只 有一个元素与之对应。 2mapPartitions mapPartitions是map的一个变种。map的输入函数是应用 于RDD中每个元素，而mapPartitions的输入函数是应用于 每个分区，也就是把每个分区中的内容作为整体来处理 的。 3mapValues mapValues顾名思义就是输入函数应用于RDD中Kev-Value 的Value，原RDD中的Key保持不变，与
11、新的Value一起组 成新的RDD中的元素。因此，该函数只适用于元素为KV对 的RDD。 4mapWith mapWith是map的另外一个变种，map只需要一个输入函数 ，而mapWith有两个输入函数。P123第8章开源集群计算环境Spark 内容CONTENTS导航Spark接口实战Spark编程之RDDSpark简介Spark MLlib实战（聚类实战）Spark实战案例之统计1000万人口的平均年龄P123第8章开源集群计算环境Spark四、Spark实战案例之统计1000万人口的平均年龄案例描述该案例中，我们将假设我们需要统计一个1000万人口的所有人的平均年龄，当然 如果想测试S
12、park对于大数据的处理能力，可以把人口数放的更大，比如1亿人口 ，当然这个取决于测试所用集群的存储容量。假设这些年龄信息都存储在一个文 件里，并且该文件的格式如下，第一列是ID，第二列是年龄。现在我们需要用Scala写一个生成1000万人口年龄数据的文件，源程序如下： 直接创建scala脚本使用上述代码运行即可获得随机生成的1000万人口年龄数据文 件。 vim SampleAge.scala scala SampleAge.scala运行成功后获得文件sample_age_data.txt。P123第8章开源集群计算环境Spark四、Spark实战案例之统计1000万人口的平均年龄案例描述
13、import java.io.FileWriter import java.io.File import scala.util.Randomobject SampleAgedef main(args:ArrayString) val writer = new FileWriter(new File(“/home/hadoop/sample_age_data.txt“),false)val rand = new Random()for ( i line.split(“ “)(1)val totalAge = ageData.map(age = Integer.parseInt(String.va
14、lueOf(age).collect().reduce(a, b) = a+b)println(“Total Age:“ + totalAge + “;Number of People:“ + count )val avgAge : Double = totalAge.toDouble / count.toDoubleprintln(“Average Age is “ + avgAge) P123第8章开源集群计算环境Spark四、Spark实战案例之统计1000万人口的平均年龄提交到集群运行要执行本实例的程序，需要将刚刚生成的年龄信息文件上传到HDFS上 运行一下HDFS命令把文件拷贝到HD
15、FS的/user/hadoop目录。本地文件上传HDFS命令： hadoop fs -put sample_age_data.txt /user/hadoopAvgAgeCalculator 类的执行命令如下： spark-submit -master spark:/192.168.10.100:7077 -name text -class spark.AvgAgeCalculator -executor-memory 512m -total-executor-cores 2 spark.jar hdfs:/master:9000/user/hadoop/sample_age_data.txt
16、P123第8章开源集群计算环境Spark四、Spark实战案例之统计1000万人口的平均年龄监控执行状态程序正确运行，则可以在控制台看到如图所示信息。也可以到Spark Web Console去查看Job的执行状态，如下图所示。P123第8章开源集群计算环境Spark 内容CONTENTS导航Spark接口实战Spark编程之RDDSpark实战案例之统计1000万人口的平均年龄Spark简介Spark MLlib实战（聚类实战）P123第8章开源集群计算环境Spark五、Spark MLlib实战（聚类实战）算法说明聚类的核心任务是：将一组目标object划分为若干个簇，每个簇 之间的object尽可能相似，簇与簇之间的object尽可能相异。 聚类算法是数据挖掘中重要的一部分，除了最为简单的K- Means聚类算法外，比较常见的还有层次法（CURE、 CHAMELEON等）、网格算法（STI
