Spark开发指南

你须要使用自己的编译系统将你的作业，打包成一套jars文件。比如，假设你使用sbt，那么sbt-assembly插件是一个好方法。将你的代码和依赖，变成一个单一的jar文件。

假设有一些类库是公用的，须要在不同的作业间共享。你可能须要手工复制到mesos的结点上，在conf/spark-env中，通过设置SPARK_CLASSPATH环境变量指向它们。

具体信息能够參考配置

分布式数据集 

Spark环绕的核心概念。是弹性分布式数据集(RDD)，一个有容错机制，能够被并行操作的集合。眼下有两种类型的RDD： 并行集合(Parrallelized Collections)，接收一个已经存在的Scala集合，在它上面执行各种并发计算； Hadoop数据集(Hadoop DataSets)，在一个文件的每条记录上，执行各种函数。仅仅要文件系统是Hdfs，或者hadoop支持的随意存储系统。

这两种RDD都能够通过同样的方式进行操作。

并行集合是通过调用SparkContext的parallelize方法。在一个已经存在的Scala集合(仅仅要是seq对象就能够)上创建而来。集合的对象将会被拷贝来创建一个分布式数据集，能够被并行操作。以下通过spark解释器的样例。展示怎样从一个数组创建一个并发集合

一旦被创建，分布数据集(distData)能够被并行操作。

比如，我们能够调用distData.reduce(_ +＿) 来将数组的元素相加。

我们会在兴许的分布式数据集做进一步描写叙述。

创建并行集合的一个重要參数，是slices的数目，它指定了将数据集切分为几份。

在集群模式中，Spark将会在一份slice上起一个Task。典型的，你能够在集群中的每一个cpu上。起2-4个Slice (也就是每一个cpu分配2-4个Task)。一般来说，Spark会尝试依据集群的状况，来自己主动设定slices的数目。

然而，你也能够手动的设置它，通过parallelize方法的第二个參数(比如：sc.parallelize(data, 10)).

Spark能够创建分布式数据集，从不论什么存储在HDFS文件系统或者Hadoop支持的其他文件系统(包含本地文件，Amazon S3， Hypertable。 HBase等等)上的文件。 Spark能够支持Text File, SequenceFiles 及其他不论什么Hadoop输入格式

文本文件的RDDs能够通过SparkContext的textFile方法创建，该方法接受文件的URI地址(或者机器上的文件本地路径，或者一个hdfs://, sdn://,kfs://,其他URI).这里是一个调用样例：

一旦被创建。distFile能够进行数据集操作。

比如。我们能够使用例如以下的map和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作业一样的方法。来设置输入源就能够了。

分布式数据集支持两种操作：

     转换(transformation):依据现有的数据集创建一个新的数据集

     动作(actions):在数据集上执行计算后，返回一个值给驱动程序

比如，Map是一个转换。将数据集的每个元素。都经过一个函数进行计算后，返回一个新的分布式数据集作为结果。而还有一方面。Reduce是一个动作，将数据集的全部元素，用某个函数进行聚合，然后将终于结果返回驱动程序。而并行的reduceByKey还是返回一个分布式数据集

全部Spark中的转换都是惰性的。也就是说。并不会立即发生计算。

相反的，它仅仅是记住应用到基础数据集上的这些转换(Transformation)。而这些转换(Transformation)，仅仅会在有一个动作(Action)发生。要求返回结果给驱动应用时，才真正进行计算。这个设计让Spark更加有效率的执行。比如，我们能够实现。通过map创建一个数据集。然后再用reduce，而仅仅返回reduce的结果给driver。而不是整个大的数据集。

spark提供的一个重要转换操作是Caching。当你cache一个分布式数据集时，每一个节点会存储该数据集的全部片，并在内存中计算。并在其他操作中重用。这将会使得兴许的计算更加的高速(一般是10倍)，缓存是spark中一个构造迭代算法的关键工具，也能够在解释器中交互使用。

以下的表格列出眼下支持的转换和动作：

转换(Transformations)

Actions(动作)

缓存

    调用RDD的cache()方法，能够让它在第一次计算后，将结果保持存储在内存。数据集的不同部分。将会被存储在计算它的不同的集群节点上。让兴许的数据集使用更快。

缓存是有容错功能的，假设任一分区的RDD数据丢失了，它会被使用原来创建它的转换，再计算一次(不须要所有又一次计算，仅仅计算丢失的分区)

Shared Variables

共享变量

一般来说，当一个函数被传递给Spark操作(比如map和reduce)。一般是在集群结点上执行。在函数中使用到的全部变量，都做分别拷贝，供函数操作，而不会互相影响。这些变量会被复制到每一台机器，而在远程机器上。在对变量的全部更新。都不会被传播回Driver程序。然而，Spark提供两种有限的共享变量，供两种公用的使用模式：广播变量和累加器

广播变量

广播变量同意程序猿保留一个仅仅读的变量，缓存在每一台机器上。而非每一个任务保存一份拷贝。他们能够使用，比如，给每一个结点一个大的输入数据集。以一种高效的方式。Spark也会尝试，使用一种高效的广播算法。来降低沟通的损耗。

    广播变量是从变量V创建的，通过调用SparkContext.broadcast(v)方法。这个广播变量是一个v的分装器，它的仅仅能够通过调用value方法获得。

例如以下的解释器模块展示了怎样应用：

在广播变量被创建后，它能在集群执行的不论什么函数上，被代替v值进行调用。从而v值不须要被再次传递到这些结点上。另外。对象v不能在被广播后改动，是仅仅读的。从而保证全部结点的变量，收到的都是一模一样的。

累加器

累加器是仅仅能通过组合操作“加”起来的变量，能够高效的被并行支持。他们能够用来实现计数器(如同MapReduce中)和求和。

Spark原生就支持Int和Double类型的计数器，程序猿能够加入新的类型。

一个计数器。能够通过调用SparkContext.accumulator(V)方法来创建。

执行在集群上的任务。能够使用+=来加值。然而，它们不能读取计数器的值。当Driver程序须要读取值的时候，它能够使用.value方法。

例如以下的解释器。展示了怎样利用累加器，将一个数组里面的全部元素相加