1、最基本的Map用法

val data = Array(1, 2, 3, 4, 5)

val distData = sc.parallelize(data)

val result = distData.map(x=>x*x)

println(result.collect().mkString(","))

其中最关键的操作就是：从分布式数据集 --转换--> 并行数据集

from a distributed dataset to Parallelized collections

Spark分布式数据集包含：

• local file system
• HDFS
• Cassandra
• HBase
• Amazon S3

Spark supports text files, SequenceFiles, and any other Hadoop InputFormat.

• 比如文件：val distFile = sc.textFile("data.txt")
• 比如hdfs：hdfs://
• 比如s3：s3n://

读取文件时需要注意的是:

• 如果使用的是本地文件路径,那么worker节点一定是有访问权限的.
• 文本文件的访问方式： textFile("/my/directory"), textFile("/my/directory/*.txt"), and textFile("/my/directory/*.gz").
• hdfs系统会把文件按128MB进行分区

2、从外部文件系统获取数据

val lines = sc.textFile("file:///usr/local/spark/examples/src/main/resources/people.json")

val lineLengths = lines.map(s => s.length)

val totalLength = lineLengths.reduce((a, b) => a + b)

3、flatMap用法

flatMap的做法有点象把迭代器拍扁拍碎，比如以下代码

val lines = sc.parallelize(List("hi man","ha girl"))

val wordsFlatmap = lines.flatMap(line=>line.split(" "))

val wordsMap = lines.map(line=>line.split(" "))

看看区别：

scala> wordsFlatmap.first

res9: String = hi

scala> wordsMap.first

res10: Array[String] = Array(hi, man)

实际上wordsFlatmap被拆成了4个string，而wordsMap是对输入的list每个元素进行了split操作，所以说flatMap的做法有点象把迭代器拍扁拍碎。比如说分词就容易用flatMap

4、笛卡尔积

在推荐系统中，要计算各用户对多个产品的兴趣度，就可以制作一个笛卡尔积，用于比较用户的的喜爱产品的相似度。

val man = sc.parallelize(List("Tom","Cat"))

val product = sc.parallelize(List("car","iphone","android","surfacePro"))

val result = man.cartesian(product)

result.collect

运行结果：

res0: Array[(String, String)] = Array((Tom,car), (Tom,iphone), (Tom,android), (Tom,surfacePro), (Cat,car), (Cat,iphone), (Cat,android), (Cat,surfacePro))

笛卡儿计算是很恐怖的，它会迅速消耗大量的内存，所以在使用这个函数的时候请小心

5、cache操作

在spark中使用cache是非常重要的，因为行动操作都是惰性求值，每次都会重新计算所有的依赖，如果有大量迭代，代价巨大。

缓存就可以从内容读取，无需再次计算

scala> var data = sc.parallelize(List(1,2,3,4))

data: org.apache.spark.rdd.RDD[Int] =

　　ParallelCollectionRDD[44] at parallelize at <console>:12

scala> data.getStorageLevel

res65: org.apache.spark.storage.StorageLevel =

　　StorageLevel(false, false, false, false, 1)

scala> data.cache

res66: org.apache.spark.rdd.RDD[Int] =

　　ParallelCollectionRDD[44] at parallelize at <console>:12

scala> data.getStorageLevel

res67: org.apache.spark.storage.StorageLevel =

　　StorageLevel(false, true, false, true, 1)

我们先是定义了一个RDD，然后通过getStorageLevel函数得到该RDD的默认存储级别，这里是NONE。然后我们调用cache函数，将RDD的存储级别改成了MEMORY_ONLY(看StorageLevel的第二个参数)

6、检查点

将生成的RDD保存到外部可靠的存储当中，对于一些数据跨度为多个bactch的有状态tranformation操作来说，checkpoint非常有必要，因为在这些transformation操作生成的RDD对前一RDD有依赖，随着时间的增加，依赖链可能会非常长，checkpoint机制能够切断依赖链，将中间的RDD周期性地checkpoint到可靠存储当中，从而在出错时可以直接从checkpoint点恢复。

val data = sc.parallelize(1 to 100 , 5)

sc.setCheckpointDir("/myCheckPoint")

data.checkpoint

data.count

7、cogroup组合

将多个RDD中同一个Key对应的Value组合到一起。

scala> val data1 = sc.parallelize(List((1, "www"), (2, "bbs")))

scala> val data2 = sc.parallelize(List((1, "iteblog"), (2, "iteblog"), (3, "very")))

scala> val data3 = sc.parallelize(List((1, "com"), (2, "com"), (3, "good")))

scala> val result = data1.cogroup(data2, data3)

scala> result.collect

res30: Array[(Int, (Iterable[String], Iterable[String], Iterable[String]))] =

Array((1,(CompactBuffer(www),CompactBuffer(iteblog),CompactBuffer(com))),

(2,(CompactBuffer(bbs),CompactBuffer(iteblog),CompactBuffer(com))),

(3,(CompactBuffer(),CompactBuffer(very),CompactBuffer(good))))

8、广播变量

广播变量是通过调用sparkcontext从变量v创建。广播变量是V的包装器，它的值可以通过调用值方法来访问。下面的代码显示了这一点：

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)

scala> broadcastVar.value

res0: Array[Int] = Array(1, 2, 3)

在创建广播变量之后，应该使用它在集群上运行的任何函数中代替V值，这样v就不会不止一次地发送到节点。此外，对象v在广播之后不应该被修改，以确保所有节点获得相同的广播变量值（例如，如果变量稍后被运送到新节点）。

9、累加器

累加器一般用来累计和计数

val accum = sc.longAccumulator("My Accumulator")

//计数

sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(1))

accum.value

res1: Long = 4

//累加

sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum.add(x))

accum.value

res2: Long = 10