ScalaInAction 数据预处理

原创
2015/10/27 17:14
阅读数 314

ScalaInAction 数据预处理

前言

使用 Spark+Scala 进行数据预处理,最大的优势就是可以处理大数据量而且速度还会很快。
Scala 不仅拥有类似与R语言一样的语法特点,还比R语言更加灵活,可以开发自己想要的工具。
Spark 基于 Scala 开发,虽然在大数据处理的时候某些语法和 List 有些不同,但是使用起来也是非常顺手。

数据源

下载测试数据集合,解压缩,数据来源是:UC Irvine Machine Learning Repository,这个里面有很多好东西。

$ mkdir linkage
$ cd linkage/
$ curl -o donation.zip http://bit.ly/1Aoywaq
$ unzip donation.zip
$ unzip 'block_*.zip'

如果有 Spark 集群可以将数据上传到集群上面,下面的测试代码只需要修改两个地方就可以运行。

数据清洗

首先数据集合有10个文件,每个文件都是 Table 结果,列按照逗号分割,而且每个文件都有 header,空值使用 ? 表示。下面要做如下几件事:

  1. 去掉 header
  def isHead(line: String): Boolean = {
    line.contains("id_1")
  }
  1. 逗号分割转换 ? 为 Double.NaN 并且保存列到指定的对象
  case class MatchData(id1: Int, id2: Int, scores: Array[Double], matched: Boolean)
  def toDouble(s: String) = {
    if ("?".equals(s)) Double.NaN else s.toDouble
  }

  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(x => toDouble(x))
    val matched = pieces(11).toBoolean
    MatchData(id1, id2, scores, matched)
  }

RDD[Double] 基本统计

这个转换的过程是比较有技术含量的,必须新建一个类,代替原来的 StatCounter 类,因为原来的 StatCounter 类不考虑空值。新建 DoubleNaNStatCounter

class DoubleNaNStatCounter extends Serializable {
  val stats: StatCounter = new StatCounter()
  var nan: Long = 0
  def add(x: Double): DoubleNaNStatCount = {
    if (java.lang.Double.isNaN(x))
      nan += 1
    else
      stats.merge(x)
    this
  }

  def merge(other: DoubleNaNStatCount): DoubleNaNStatCount = {
    stats.merge(other.stats)
    nan += other.nan
    this
  }

  override def toString = {
    "stats: " + stats.toString() + " NaN: " + nan
  }
}

object DoubleNaNStatCount extends Serializable {
  def apply(x: Double) = new DoubleNaNStatCount().add(x)
}

再加一个辅助方法,这个方法非常重要,partition本地处理,减少数据传输,优化效率:

  def statsWithMissing(rdd: RDD[Array[Double]]): Array[DoubleNaNStatCount] = {
    val nastats = rdd.mapPartitions((iter: Iterator[Array[Double]]) => {
      val nas: Array[DoubleNaNStatCount] = iter.next().map(d => DoubleNaNStatCount(d))
      iter.foreach(arr => {
        nas.zip(arr).foreach { case (n, d) => n.add(d) }
      })
      Iterator(nas)
    })
    nastats.reduce((n1, n2) => {
      n1.zip(n2).map { case (a, b) => a.merge(b) }
    })
  }

上面类和方法看懂了,基本对 Spark 工作原理就懂了。上面两个操作,等于自己实现了一个更加通用的 Stat,可以作为以后的工作方法用。

驱动代码

  def main(args: Array[String]) {
    //master指定为本地,意味着这是测试
    val conf = new SparkConf().setAppName("SparkInAction").setMaster("local[4]")
    val sc = new SparkContext(conf)
    //如果不是测试这个路径修改为对应的正确位置
    val rdd = sc.textFile("F:\\clebeg\\spark\\donation")
    //去除每个文件的头信息
    val noHeader = rdd.filter(!isHead(_))
    val parsed = noHeader.map(parse)
    //如何转换成Map,如何为Map排序
    val matchCount = parsed.map(md => md.matched).countByValue()
    matchCount.foreach(println)
    matchCount.toSeq.sortBy(_._1).foreach(println)
    matchCount.toSeq.sortBy(_._2).foreach(println)
    //RDD[Double] 通过隐式类型转换具有 stats 方法

    //下面查看匹配和不匹配的数据之间的差异
    val nasm = statsWithMissing(parsed.filter(_.matched).map(_.scores))
    val nasn = statsWithMissing(parsed.filter(!_.matched).map(_.scores))
    val diff = nasm.zip(nasn).map{case (a, b) => (a.nan + b.nan, a.stats.mean - b.stats.mean)}
    diff.foreach(println)
  )
展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部