转载必须注明出处：梁杰帆
在这里要先感谢原作者们！如果各位在这里发现了错误之处，请大家提出
1.Initializing Spark
Spark程序必须做的第一件事就是创建一个SparkContext对象，它告诉Spark如何访问集群。要创建SparkContext，首先需要构建一个SparkConf对象，该对象包含关于应用程序的信息。
val conf = new SparkConf().setAppName(appName).setMaster(master)
val scc = new StreamingContext(conf) (比如创建StreamingContext)
2.Using the Shell
1.一般性:
./bin/spark-shell --master local[4]
2.添加jar包路径:
./bin/spark-shell --master local[4] --jars code.jar
3.包含Maven依赖资源:
./bin/spark-shell --master local[4] --packages "org.example:example:0.1"
3.Resilient Distributed Datasets(RDDs)--弹性分布式数据集
它是spark中的一个抽象，是一个数据集合，能够被分布式系统并行操作。有两种方法创建RDD，一是并行驱动程序现有的数据集；二是在外部存储系统中引用数据集，例如共享的文件系统、HDFS、HBase或任何提供Hadoop InputFormat的数据源。
1.Parallelized Collections--并行化数据集处理
1.使用 SparkContext’s parallelize 接口在程序中把已有的集合转换成RDD供分布式并行操作。例如：
val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)
distData.reduce((a, b) => a + b)--call by name
2.并行化RDD时，一个partitions对应一个task，spark会自动的根据集群来使用几个分区来进行并行化。不过也可人为指定partitions的数量。例如：
sc.parallelize(data, 10)
2.External Datasets
1.读取外部文件系统作为RDD，然后直接可进行分布式数据操作例如：
val distFile = sc.textFile("data.txt")
distFile.map(s => s.length).reduce((a, b) => a + b)
2.如果读取的是本地文件系统上的文件，使用路径必须在工作节点上的相同路径上访问该文件。要么将文件复制到所有工作节点，要么使用网络挂载共享文件系统。即读取本地系统的文件，集群上所有的节点都必须有这个路径的文件存在。
3.Spark的所有基于文件的输入方法，包括textFile，支持在目录、压缩文件和通配符上运行，例如：
textFile("/my/directory")
textFile("/my/directory/*.txt")
textFile("/my/directory/*.gz")
4.spark一般是一个分区对应一个block(针对hdfs的block，一般是128M)，数据量大时，也可人为设定控制分区数，加快读取速度和并行化成RDD数据。
5.除了textfile，spark还支持以下的输入接口：
1.读取一个目录的很多小文件，每个文件中返回一条记录。分区是由数据位置决定的，在某些情况下，可能导致分区太少。对于这些情况，纯文本文件提供了一个可选的第二个参数来控制最小的分区数。
SparkContext.wholeTextFiles
2.读取二进制类的文件(视频，音乐，图片等资源)， 使用SparkContext.sequenceFile[K, V]中的K V要符合hadoop的Writeable接口的定义：
SparkContext.SequenceFiles
3.读取hadoop inputformat的文件，像hadoop读取输入源一样使用:
SparkContext.hadoopRDD(old API) / SparkContext.newAPIHadoopRDD(new API)
4.一种简单的格式(由序列化的Java对象组成)来保存RDD。虽然这并不像Avro那样的特殊格式，但它提供了一种简单的方法来保存任何RDD:
SparkContext.RDD.saveAsObjectFile / SparkContext.objectFile
3.RDD Operations
1.RDD支持两种操作：
1.从一个已存在的dataset转换成一个新的dataset
transformations，所有的transformations都是lazy(延迟估值，即直至在actions前一刻都只是记录转换后的dataset，不涉及运算actions)
2.对RDD进行一系列的操作，cache等持久化RDD操作可优化运算速度，有预测读作用和把经常读的和即将读的持久化到内存，如用于ML的迭代运算
actions
例如：
val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
第一行通过外部文件定义了一个基本RDD。该数据集没有装入内存或其他操作:lines仅仅是文件的指针。第二行定义了lineLengths作为map转换的结果。由于懒惰，lineLengths并没有立即被计算出来。最后，运行reduce，这是一个actions。这一刻， Spark将计算分解成在独立的机器上运行的任务，并且每台机器都运行它的部分map和局部还原，只返回其对驱动程序的结果。
val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
lineLengths.persist()---固化linesLengths在内存，供再次使用
val totalLength = lineLengths.reduce((a, b) => a + b)
2.Passing Functions to Spark
1.Spark的API严重依赖于驱动程序中的传递函数在集群上运行。有两种推荐方法:
1.匿名函数语法，可用于短代码。
2.全局单例对象中的静态方法，如下:
object MyFunctions {
def func1(s: String): String = { ... }
}
myRdd.map(MyFunctions.func1)
3.虽然在类实例中传递对方法的引用也是可能的，但这需要发送包含该类的对象和方法。例如:
class MyClass {
def func1(s: String): String = { ... }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}
在这里，如果我们创建一个新的MyClass实例并在它上面调用doStuff，那么其中的映射将引用MyClass实例的func1方法，因此需要将整个对象发送到集群。它类似于编写rdd.masp(x = > this.func1(x))。
4.以类似的方式，访问外部对象的字段将引用整个对象:
class MyClass {
val field = "Hello"
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field_ + x) }
}
5.相当于写rdd.map(x = > this.field+ x)，它引用了所有this。为了避免这个问题，最简单的方法是将field复制到本地变量中，而不是从外部访问它:
def doStuff(rdd: RDD[String]): RDD[String] = {
val field_ = this.field
rdd.map(x => field_ + x)
}
3.Understanding closures--理解闭包
1.闭包发送给每个executor的变量都是副本，当counter在foreach函数中被引用时，它不再是驱动节点上的计数器。在driver的内存中仍然有一个计数器，但是executor不可见的。executor只能看到序列化的闭包中的副本。因此，计数器的最终值仍然为零，因为 counter上的所有操作都引用了序列化闭包中的值
2.在本地模式中，在某些情况，foreach函数实际上会在与driver相同的JVM中执行，并引用相同的原始计数器，并可能实际更新它。
3.当需要在集群中累加更新变量，需要用到 Accumulator-累加器。系统框架帮我们保护 Accumulator 。
4。闭包像循环或局部定义的方法，不应该用来改变一些全局状态。Spark没有定义或保证从闭包外部引用的对象的突变行为。有些代码可以在本地模式下工作，但这只是偶然的，放在分布式模式可能不正确。如果需要一些全局聚合，使用累加器
4.Printing elements of an RDD---打印RDD的元素
1.rdd.foreach(println) or rdd.map(println)-----只适用本地模式打印RDD元素
2.rdd.collect().foreach(println)-----适合集群。原因是，在集群中，foreach是讯轮所有的元素来打印，但一个RDD中一般没有包含所有的元素。
3。take(): rdd.take(100).foreach(println) 用于打印一部分RDD元素
5.Working with Key-Value Pairs
1.大多数Spark操作都在包含任何类型对象的RDDs上工作，但一些特殊操作只能在键值对的RDDs上使用。最常见的是分布式shuffle操作，例如按一个键分组或聚合元素。
2.当使用自定义对象作为键-值对操作的关键时，您必须确保自定义equals()方法附带一个匹配的hashCode()方法。详细用法在Object.hashCode() documentation。
6.Transformations
Transformation Meaning
map(func) 作用每个元素，映射1个输出项，1对1
filter(func) 过滤元素。1对1
flatMap(func) 作用每个输入项，映射1个set。1对n
mapPartitions(func) 作用一个RDD输出n个分区，Iterator<T>(old) => Iterator<U>(new)
mapPartitionsWithIndex(func) 类似 mapPartitions ，但提供分区索引
union(otherDataset) 把多个RDD合成成一个新的RDD
sample(withReplacement, fraction, seed) 根据随机种子，是否替换数据的一部分数据
intersection(otherDataset) 返回一个新的RDD，包含源数据集中元素和参数的交集
distinct([numTasks])) 返回包含源数据集不同元素的新数据集
groupByKey([numTasks]) 返回分组的RDD，(K, V) pairs => (K, Iterable<V>) pairs.
(默认情况下，输出的并行度取决于父RDD的分区数。您可以通过一个可选的numTasks参数来设置不同数量的任务)
reduceByKey(func, [numTasks]) 返回按key统计的RDD，(K,V)pairs=>(K,V)pairs (V,V)=>V.Like in groupByKey
aggregateByKey(zeroValue)(seqOp, combOp, [numTasks]) (K, V)pairs=>(K, U)pairs 每个键的值都与给定的组合函数和中立的“零”值聚合
sortByKey([ascending], [numTasks]) 对RDD所有的元素按key进行排序
join(otherDataset, [numTasks]) 对两个RDD进行join操作
cogroup(otherDataset, [numTasks]) (K, V) and (K, W), 返回(K, (Iterable<V>, Iterable<W>)) tuples
cartesian(otherDataset) 笛卡儿操作，数据集T 和 数据集U, 返回数据集(T, U) pairs
pipe(command, [envVars]) 通过shell将RDD的每个分区建管道。将RDD元素写入stdin，将输出重定向stdout作为字符串RDD返回
coalesce(numPartitions) 将RDD中的分区数量减少到num分区。在过滤大数据集后更有效地运行操作
repartition(numPartitions) 在RDD中随机重组数据，创建更多或更少的分区，重新负载均衡。
repartitionAndSortWithinPartitions(partitioner) 根据给定的分区器重新划分RDD，并在每个结果分区中按其键对记录进行排序
7.Actions
Action Meaning
reduce(func) 使用函数func聚合数据集的元素(它需要两个参数并返回一个参数)
collect() 将数据集的所有元素作为数组在驱动程序中返回。用于在筛选或其他操作返回足够小的数据子集后
count() 统计数据集的元素个数
first() 返回数据集第一个元素
take(n) 返回数据集指定元素
takeSample(withReplacement, num, [seed]) 返回一个数组，通过随机种子选出部分元素组成数组
takeOrdered(n, [ordering]) 使用自然顺序或自定义比较器返回RDD的前n个元素
saveAsTextFile(path) 将数据集的元素作为文本文件写入本地文件系统、HDFS或任何其他hadoop支持文件系统的指定目录中。
Spark将在每个元素上调用toString，将其转换为文件中的一行文本。
saveAsSequenceFile(path) 将数据集的元素作为序列文件写入本地文件系统、HDFS或任何其他hadoop支持文件系统的指定目录中。
saveAsObjectFile(path) 使用Java序列化的简单格式编写数据集的元素，然后可以使用SparkContext.objectFile()加载。
countByKey() 只有在类型(K,V)类型的RDDs上才可用。返回一个hashmap(K,Int)对每个键的计数
foreach(func) 轮训访问RDD中每一个元素，注意在累加器之外的使用，注意闭包
8.Shuffle operations
1.Background
1.Spark中的某些操作触发一个称为shuffle的事件。shuffle是Spark的重新分发数据的机制，因此它在不同的分区上有不同的分组。通常涉及在executors和机器之间复制数据，从而使shuffle操作变得复杂且代价高昂，是优化的重点地方
2.在Spark中，数据一般不会跨分区分布到需要特定操作的地方。在计算过程中，单个任务将在单个分区上操作——因此，为了组织所有的数据，以减少单个的关键任务来执行，Spark需要执行一个全面的操作。它必须从所有分区读取所有键的值，然后将各个分区的值集合起来，以计算每个键的最终结果——这称为shuffle
3.shffle后的分区中的元素是确定的，分区本身的顺序也是确定的。但分区里面的元素不是有序的。要想有序，先执行以下的操作：
mapPartitions 执行分区排序曹组, 例如.sorted
repartitionAndSortWithinPartitions 在同时重新分区的同时，有效地对分区进行排序
sortBy 使用全局的RDD
4.可以引起shuffle的操作
repartition、coalesce、groupByKey、reduceByKey、join、cogroup
2.Performance Impact---性能影响
1.当数据不适合内存时，Spark将把这些表溢写到磁盘上，从而增加磁盘I/O的额外开销并增加垃圾收集
2.Shuffle还在磁盘上生成大量的中间文件。在Spark 1.3中，这些文件被保存，直到相应的RDDs不再被使用，并且被垃圾收集。这已经完成了，因此如果重新计算了沿袭，则不需要重新创建shuffle文件。如果应用程序保留对这些RDDs的引用，或者GC不频繁地启动，则垃圾收集可能会在很长一段时间后才会发生。这意味着长时间运行的Spark作业可能消耗大量磁盘空间。临时存储目录由 spark.local.dir指定。
3.Shuffle的行为可以通过调整各种配置参数来调整
4.RDD Persistence---RDD持久化
1.RDD持久化在内存中，可以使后面的actions加快运行，缓存是迭代算法和快速交互使用的关键工具。第一次RDD调用persist() or cache()持久化RDD到节点的内存，Spark的缓存是容错的——如果RDD的任何分区丢失，则它将使用最初创建的转换自动重新计算。利用的RDD特性(RDD是无状态)
2.每个RDD持久化到不同的内存。
MEMORY_ONLY 仅持久化到内存
MEMORY_AND_DISK 持久化到内存和磁盘
MEMORY_ONLY_SER 将RDD存储为序列化的Java对象(每个分区的一个字节数组)
MEMORY_AND_DISK_SER 持久化到内存、磁盘、Java对象
DISK_ONLY 仅持久化到磁盘
MEMORY_ONLY_2, MEMORY_AND_DISK_2, 同上，不过复制到两个节点
OFF_HEAP 将数据存储在非堆内存中。这需要启用非堆内存
3.Which Storage Level to Choose?
1.如果RDDs与默认的存储级别(MEMORY_ONLY)适应，使用此方式。这是最有效的，允许RDDs上的操作尽可能快地运行。
2.如果不符合1，尝试用MEMORY_ONLY_SER并选择一个快速序列化库，以使对象更具有空间效率，但仍可以相当快地访问。
3.不要溢出到磁盘，除非计算数据集的函数是昂贵的，或者它们过滤大量的数据。否则，重新计算分区可能和从磁盘读取它一样快。
4.选择顺序：MEMORY_ONLY--> MEMORY_ONLY_SER --> DISK_ONLY
5.快速的故障恢复(例如，如果使用Spark服务来自web应用程序的请求)，应使用复制存储级别。所有的存储级别都通过重新计算丢失的数据来提供完全的容错，但是复制的数据允许继续在RDD上运行任务，无需等待重新计算丢失的分区。
4.Removing Data
1.Spark自动监视每个节点上的缓存使用情况，并在最近使用的(LRU)中删除旧的数据分区。也可手动删除一个RDD，而不是等待它从缓存中退出，使用RDD.unpersist()方法。
4.Shared Variables---共享变量
1.Broadcast Variables---广播变量
1.广播变量允许将只读变量保存在每台机器上，而不是将其复制到任务中。例如，可以使用它们以有效的方式为每个节点提供一个大的输入数据集的副本。Spark还尝试使用高效的广播算法来分发广播变量，以减少通信成本。
2.Sparkactions通过一组stages执行，通过分布式“shuffle”操作分隔。Spark将在每个阶段自动地广播任务所需的公共数据。在运行每个任务之前，以序列化的形式缓存该方法的数据并进行反序列化。意味着，显式创建广播变量仅对当跨多个阶段的任务需要相同的数据时，或者当以反序列化形式缓存数据时有用。
2.创建 Broadcast Variables 例子：(广播变量是一个关于v的包装器，它的值可以通过调用值方法来访问)
val broadcastVar = SparkContext.broadcast(Array(1, 2, 3))
2.Accumulators---累加器
1.Accumulators默认支持数字类型的累加器，但可自定义类型
2.能精确地统计数据的各种属性。例如可以统计符合userID的记录数，在某时间段内产生的购买次数,在ETL使用Accumulator去统计出各种属性的数据
3.轻量级的调试工具，在UI中跟踪累加器，对于了解运行阶段的进度非常有用
4.从集群的资源利用率来精确的测量出Spark应用的资源利用率
5.创建数字Accumulators
SparkContext.longAccumulator() or SparkContext.doubleAccumulator()
6.自定义Accumulators
class VectorAccumulatorV2 extends AccumulatorV2[MyVector, MyVector] {
private val myVector: MyVector = MyVector.createZeroVector
def reset(): Unit = {
myVector.reset()
}
def add(v: MyVector): Unit = {
myVector.add(v)
}
...
}
val myVectorAcc = new VectorAccumulatorV2 //创建一个累加器
sc.register(myVectorAcc, "MyVectorAcc1") //注册累加器到spark上下文
7.对于只在操作中执行的累加器更新，Spark保证每个任务对累加器的更新只会被应用一次，即重新启动的任务不会更新该值。在Transformations中，应意识到，如果任务或工作阶段被重新执行，每个任务的更新可能不止一次。
8.累加器不能改变Spark的lazy模型。如果它们在RDD的操作中被更新，那么它们的值只有在RDD作为actions的一部分被计算时才更新。因此，在像map()这样的惰性Transformations中，不保证会执行累加器更新。
参考： http://spark.apache.org/docs/latest/rdd-programming-guide.html