Spark基础脚本入门实践2:基础开发

时间:2023-03-08 16:17:41
Spark基础脚本入门实践2:基础开发

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

Spark基础脚本入门实践2:基础开发

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、累加器

Spark基础脚本入门实践2:基础开发
累加器一般用来累计和计数
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