SPARK 开发指南
- 格式：pdf
- 大小：237.99 KB
- 文档页数：9
Spark开发指南
从高的层面来看，其实每一个Spark的应用，都是一个Driver类，通过运行用户定义的main 函数，在集群上执行各种并发操作和计算
Spark提供的最主要的抽象，是一个弹性分布式数据集(RDD)，它是一种特殊集合，可以分布在集群的节点上，以函数式编程操作集合的方式，进行各种各样的并发操作。
它可以由hdfs上的一个文件创建而来，或者是Driver程序中，从一个已经存在的集合转换而来。
用户可以将数据集缓存在内存中，让它被有效的重用，进行并发操作。
最后，分布式数据集可以自动的从结点失败中恢复，再次进行计算。
Spark的第二个抽象，是并行计算中使用的共享变量。
默认来说，当Spark并发运行一个函数时，它是以多个的task，在不同的结点上运行，它传递每一个变量的一个拷贝，到每一个独立task使用到的函数中，因此这些变量并非共享的。
然而有时候，我们需要在任务中能够被共享的变量，或者在任务与驱动程序之间共享。
Spark支持两种类型的共享变量：广播变量：可以在内存的所有结点中被访问，用于缓存变量（只读）累加器：只能用来做加法的变量，例如计数和求和
本指南通过一些样例展示这些特征。
读者最好是熟悉Scala，尤其是闭包的语法。
请留意，Spark可以通过Spark-Shell的解释器进行交互式运行。
你可能会需要它。
注：所谓“闭包”，指的是一个拥有许多变量和绑定了这些变量的环境的表达式（通常是一个函数），因而这些变量也是该表达式的一部分。
接入Spark
为了写一个Spark的应用，你需要将Spark和它的依赖，加入到CLASSPATH中。
最简单的方法，就是运行sbt/sbt assembly来编译Spark和它的依赖，打到一个Jar里面
core/target/scala_2.9.1/spark-core-assembly-0.0.0.jar，然后将它加入到你的CLASSPATH 中。
或者你可以选择将spark发布到maven的本地缓存中，使用sbt/sbt publish。
它将在组织org.spark-project下成为一个spark-core.
另外，你会需要导入一些Spark的类和隐式转换，将下面几行加入到你程序的顶部
import spark.SparkContext
import SparkContext._
初始化Spark
distData:spark.RDD[Int]=spark.ParallelCollection@10d13e3e
一旦被创建，分布数据集（distData）可以被并行操作。
例如，我们可以调用distData.reduce(_ +＿)来将数组的元素相加。
我们会在后续的分布式数据集做进一步描述。
创建并行集合的一个重要参数，是partition的数目，它指定了将数据集切分为几份。
在集群模式中，Spark将会在一份partition上起一个Task。
典型的，你可以在集群中的每个cpu 上，起2-4个partitions（也就是每个cpu分配2-4个Task）。
一般来说，Spark会尝试根据集群的状况，来自动设定partitions的数目。
然而，你也可以手动的设置它，通过parallelize 方法的第二个参数（例如：sc.parallelize(data,10)).
Hadoop数据集
Spark可以创建分布式数据集，从任何存储在HDFS文件系统或者Hadoop支持的其它文件系统（包括本地文件，Amazon S3，Hypertable，HBase等等）上的文件。
Spark可以支持Text File,SequenceFiles及其它任何Hadoop输入格式
文本文件的RDDs可以通过SparkContext的textFile方法创建，该方法接受文件的URI地址（或者机器上的文件本地路径，或者一个hdfs://,sdn://,kfs://,其它URI).这里是一个调用例子：
scala>val distFile=sc.textFile(“data.txt”)
distFile:spark.RDD[String]=spark.HadoopRDD@1d4cee08
一旦被创建，distFile可以进行数据集操作。
例如，我们可以使用如下的map和reduce操作将所有行数的长度相加：
distFile.map(_.size).reduce(_+_)
方法也接受可选的第二参数，来控制文件的分片数目。
默认来说，Spark为每一块文件创建一个分片（HDFS默认的块大小为64MB)，但是你可以通过传入一个更大的值来指定更多的分片。
注意，你不能指定一个比块个数更少的片值（和hadoop中，Map数不能小于Block 数一样）
对于SequenceFiles，使用SparkContext的sequenceFile[K,V]方法，K和V是文件中的key 和values类型。
他们必须是Hadoop的Writable的子类，例如IntWritable和Text。
另外，Spark允许你指定几种原生的通用Writable类型，例如：sequencFile[Int,String]会自动读取IntWritable和Texts
最后，对于其他类型的Hadoop输入格式，你可以使用SparkContext.hadoopRDD方法，它可以接收任意类型的JobConf和输入格式类，键类型和值类型。
按照对Hadoop作业一样的方法，来设置输入源就可以了。
RDD的持久化
用法:使用persist()或者cache()方法,其中cache()方法默认持久化到内存,persist可以自己选择持久化的层次,在shuffle操作中,spark会自动保存中间计算结果,例如reduceBykey
广播变量
广播变量允许程序员保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份拷贝。
他们可以使用，例如，给每个结点一个大的输入数据集，以一种高效的方式。
Spark也会尝试，使用一种高效的广播算法，来减少沟通的损耗。
广播变量是从变量V创建的，通过调用SparkContext.broadcast(v)方法。
这个广播变量是一个v的分装器，它的只可以通过调用value方法获得。
如下的解释器模块展示了如何应用：
scala>val broadcastVar=sc.broadcast(Array(1,2,3))
broadcastVar:spark.Broadcast[Array[Int]]=
spark.Broadcast(b5c40191-a864-4c7d-b9bf-d87e1a4e787c)
scala>broadcastVar.value
res0:Array[Int]=Array(1,2,3)
在广播变量被创建后，它能在集群运行的任何函数上，被取代v值进行调用，从而v值不需要被再次传递到这些结点上。
另外，对象v不能在被广播后修改，是只读的，从而保证所有结点的变量，收到的都是一模一样的。
累加器
累加器是只能通过组合操作“加”起来的变量，可以高效的被并行支持。
他们可以用来实现计数器（如同MapReduce中）和求和。
Spark原生就支持Int和Double类型的计数器，程序员可以添加新的类型。
一个计数器，可以通过调用SparkContext.accumulator(V)方法来创建。
运行在集群上的任务，可以使用+=来加值。
然而，它们不能读取计数器的值。
当Driver程序需要读取值的时候，它可以使用.value方法。
如下的解释器，展示了如何利用累加器，将一个数组里面的所有元素相加
scala>val accum=sc.accumulator(0)
accum:spark.Accumulator[Int]=0
scala>sc.parallelize(Array(1,2,3,4)).foreach(x=>accum+=x)
…
10/09/2918:41:08INFO SparkContext:Tasks finished in0.317106s。
