Spark数据分析之第2课

本课主要介绍对类别变量基数相对小的数据创建直方图，比如本课中对匹配和不匹配的记录数统计。

#读取HDFS文件并创建RDD

scala> val rawblocks = sc.textFile("/linkage")

rawblocks: org.apache.spark.rdd.RDD[String] = /linkage MapPartitionsRDD[1] at textFile at <console>:27

#定义判断行中是否包含指定字符串的函数

scala> def isHeader(line:String) = line.contains("id_1")

isHeader: (line: String)Boolean

#但是通过下标访问不是很好理解，下面介绍如何通过字段名称来访问

#case class提供这种功能，case class是不可变类的一种简单类型，内置了所有Java类的基本方法，比如toString,equals和hashCode

scala> case class MatchData(id1: Int, id2: Int, scores: Array[Double], matched: Boolean)

defined class MatchData

#写一个函数处理，遇到? 时返回NaN

def toDouble(s: String) = {

    if ("?".equals(s)) Double.NaN else s.toDouble

}

#修改parse方法，以返回MatchData实例

def parse(line: String) = {

    val pieces = line.split(",")

    val id1 = pieces(0).toInt

    val id2 = pieces(1).toInt

    val scores = pieces.slice(2,11).map(toDouble)

    val matched = pieces(11).toBoolean

    MatchData(id1,id2,scores,matched)

}

#应用到/linkage目录下面的所有数据上

scala> val mds = rawblocks.filter(x => !isHeader(x)).map(x => parse(x))

mds: org.apache.spark.rdd.RDD[MatchData] = MapPartitionsRDD[3] at map at <console>:37

scala> mds.take(1)

res0: Array[MatchData] = Array(MatchData(37291,53113,[D@1d5e1c6f,true))

#下面我们针对上面的RDD进行groupBy聚合操作，获得Scala Map[Boolean,Array[MatchData]]，其中键值基于MatchData类的字段matched

scala> val grouped = mds.groupBy(md => md.matched)

grouped: org.apache.spark.rdd.RDD[(Boolean, Iterable[MatchData])] = ShuffledRDD[5] at groupBy at <console>:39

#使用mapValues得到计数

scala> grouped.mapValues(x => x.size)

res5: org.apache.spark.rdd.RDD[(Boolean, Int)] = MapPartitionsRDD[14] at mapValues at <console>:42

scala> grouped.mapValues(x => x.size).foreach(println)

(false,5728201)                                                                  

(true,20931)

#创建直方图

RDD[T]类已经定义了一个名为countByValue的动作，该动作对于计数类运算效率非常高，它向客户端返回Map[T,Long]类型的结果。

对MatchData记录中的matched字段映射调用countByValue会执行一个Spark作业，并向客户端返回结果。

scala> val nohead = rawblocks.filter(x => !isHeader(x))

nohead: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[24] at filter at <console>:31

scala> val parsed = nohead.map(line => parse(line))

parsed: org.apache.spark.rdd.RDD[MatchData] = MapPartitionsRDD[25] at map at <console>:39

scala> parsed.cache()

res17: parsed.type = MapPartitionsRDD[25] at map at <console>:39

scala> val matchCounts = parsed.map(md => md.matched).countByValue()

matchCounts: scala.collection.Map[Boolean,Long] = Map(true -> 20931, false -> 5728201)

Scala的Map类没有提供根据内容的键或值排序的方法，但是我们可以将Map转换成Scala的Seq类型，而Seq类支持排序。Scala的Seq类和Java的List接口类似，都是

可迭代集合，即具有确定的长度并且可以根据下标来查找值:

scala> val matchCountSeq = matchCounts.toSeq

matchCountSeq: Seq[(Boolean, Long)] = ArrayBuffer((true,20931), (false,5728201))

#使用sortBy方法控制排序方式，默认情况下，对数值按升序排序，通过使用序列上的reverse方法，达到降序排序

scala> matchCountSeq.sortBy(_._1).foreach(println)

(false,5728201)

(true,20931)

scala> matchCountSeq.sortBy(_._2).foreach(println)

(true,20931)

(false,5728201)

scala> matchCountSeq.sortBy(_._2).reverse.foreach(println)

(false,5728201)

(true,20931)