Spark-ML-数据获取/处理/准备

获取公开数据集

UCL机器学习知识库：包括近300个不同大小和类型的数据集，可用于分类、回归、聚类 和推荐系统任务。数据集列表位于：http://archive.ics.uci.edu/ml/。
Amazon AWS公开数据集：包含的通常是大型数据集，可通过Amazon S3访问。这些数据 集包括人类 基因组项目 、 Common Crawl 网页语料 库、维基百 科数据和 Google Books Ngrams。相关信息可参见：http://aws.amazon.com/publicdatasets/。
Kaggle：这里集合了Kaggle举行的各种机器学习竞赛所用的数据集。它们覆盖分类、回 归、排名、推荐系统以及图像分析领域，可从Competitions区域下载：http://www.kaggle.com/ competitions。
KDnuggets：这里包含一个详细的公开数据集列表，其中一些上面提到过的。该列表位 于：http://www.kdnuggets.com/datasets/index.html。
注意:
Spark 1.2.0引入了一个实验性质的新MLlib API，位于ml包下（现有的接口 则位于mllib包下）。新API旨在加强原有的API和接口的设计，从而更容易衔接 数据流程的各个环节。这些环节包括特征提取、正则化、数据集转化、模型训练 和交叉验证。
新API仍处于实现阶段，在后续的版本中可能会出现重大的变更。因此，后 续的章节将只关注相对更成熟的现有MLlib API。随着版本的更新，本书所提到 的各种特征提取方法和模型将会简单地桥接到新API中。但新API的核心思路和 大部分底层代码仍会保持原样。

探索与可视化数据

1. 用户数据
2. 电影数据
3. 评级数据

处理与转换数据

过滤掉或删除非规整或有值缺失的数据：这通常是必须的，但的确会损失这些数据里那 些好的信息。
填充非规整或缺失的数据：可以根据其他的数据来填充非规整或缺失的数据。方法包括 用零值、全局期望或中值来填充，或是根据相邻或类似的数据点来做插值（通常针对时 序数据）等。选择正确的方式并不容易，它会因数据、应用场景和个人经验而不同。
对异常值做鲁棒处理：异常值的主要问题在于即使它们是极值也不一定就是错的。到底 是对是错通常很难分辨。异常值可被移除或是填充，但的确存在某些统计技术（如鲁棒 回归）可用于处理异常值或是极值。
对可能的异常值进行转换：另一种处理异常值或极值的方法是进行转换。对那些可能存 在异常值或值域覆盖过大的特征，利用如对数或高斯核对其转换。这类转换有助于降低 变量存在的值跳跃的影响，并将非线性关系变为线性的。

从数据中提取有用特征

数值特征（numerical feature）：这些特征通常为实数或整数，比如之前例子中提到的年龄。
 类别特征（categorical feature）：它们的取值只能是可能状态集合中的某一种。我们数据 集中的用户性别、职业或电影类别便是这类。
 文本特征（text feature）：它们派生自数据中的文本内容，比如电影名、描述或是评论。
注意：

  分词（tokenization）：首先会应用某些分词方法来将文本分隔为一个由词（一般如单词、 数字等）组成的集合。可用的方法如空白分隔法。这种方法在空白处对文本分隔并可能 还删除其他如标点符号和其他非字母或数字字符。

 删除停用词（stop words removal）：之后，它通常会删除常见的单词，比如the、and和but
（这些词被称作停用词）。


  提取词干（stemming）：下一步则是词干的提取。这是指将各个词简化为其基本的形式或 者干词。常见的例子如复数变为单数（比如dogs变为dog等）。提取的方法有很多种，文本 处理算法库中常常会包括多种词干提取方法。

  向量化（vectorization）：最后一步就是用向量来表示处理好的词。二元向量可能是最为 简单的表示方式。它用1和0来分别表示是否存在某个词。从根本上说，这与之前提到的k 之1编码相同。与k之1相同，它需要一个词的字典来实现词到索引序号的映射。随着遇到 的词增多，各种词可能达数百万。由此，使用稀疏矩阵来表示就很关键。这种表示只记 录某个词是否出现过，从而节省内存和磁盘空间，以及计算时间。

 其他特征：大部分其他特征都最终表示为数值。比如图像、视频和音频可被表示为数值 数据的集合。地理位置则可由经纬度或地理散列（geohash）表示。

用软件包提取特征

Spark支持Scala、Java和Python的绑定。我们可以通过这些语言所开发的软件包，借助其中完 善的工具箱来实现特征的处理和提取，以及向量表示。特征提取可借助的软件包有scikit-learn、gensim、scikit-image、matplotlib、Python的NLTK、Java编写的OpenNLP以及用Scala编写的Breeze和Chalk。实际上，Breeze自Spark 1.0开始就成为Spark的一部分了。后几章也会介绍如何使用Breeze
的线性代数功能。

代码

MovieLens 100k数据集

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.mllib.stat.Statistics
import scala.math._
import org.apache.spark.mllib._
import org.apache.spark.sql.catalyst.expressions.Length
import org.apache.spark.mllib.feature.Normalizer
import org.apache.spark.mllib.linalg._
object Mian {
  def convertYear(x: String): Int = {
    try {
      return (x.substring(x.length()-4)).toInt
    } catch {
      case t: Throwable => return 1990
    }

  }
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("ml").setMaster("local")
    val sc = new SparkContext(conf)
    val userData = sc.textFile("data/u.user")
    //println(data.first())
    val userFields = userData.map(line => line.split("|"))
    val numUsers = userFields.map(fields => fields(0)).count()
    val numGenders = userFields.map(fields => fields(2)).distinct().count()
    val numOccupations = userFields.map(fields => fields(3)).distinct().count()
    val numZipCodes = userFields.map(fields => fields(4)).distinct().count()
    println("用户： %d,性别: %d，职业：%d,邮编：%d".format(numUsers,numGenders,numOccupations,numZipCodes))
    val movieData = sc.textFile("data/u.item")
    println("电影数量：%d".format(movieData.count()))
    val movieFields = movieData.map(line => line.split("|"))
    val years = movieFields.map(field => field(2)).map(x => convertYear(x))
    val yearsFilter = years.filter(x=>x!=1990)
    val numMovie = years.count()
    val movieAge = yearsFilter.map(y => 1998-y).countByValue()
    val rattingData = sc.textFile("data/u.data")
    val rattingFields = rattingData.map(line=>line.split("\t"))
    val rattings = rattingFields.map(field => field(2).toInt)
    val numRatting = rattings.count()
    val maxRatting = rattings.reduce((x, y) => math.max(x, y))
    val minRatting = rattings.reduce((x, y) => math.min(x, y))
    val meanRatting = rattings.reduce((x,y) => (x+y)) / numRatting
    val rattingsPerUser = numRatting / numUsers
    val rattingsPerMovie = numRatting / numMovie
    println("最小评分：",minRatting)
    println("最大评分：",maxRatting)
    println("平均评分：",meanRatting)
    println("每个用户平均评分：",rattingsPerUser)
    println("每个电影平均评分：",rattingsPerMovie)
    val userRattingGrpuped = rattingFields.map(field => (field(0).toInt,field(2).toInt)).groupByKey()
    //用户评级次数
    val userRattingByUser = userRattingGrpuped.map((k) => (k._1,k._2.size))
   // userRattingByUser.take(5)
    println(userRattingByUser.take(5).toMap)
    val allOccupations = userFields.map(fields => fields(3)).distinct().collect()
    println(allOccupations.sorted.toString())

    val normalizer = new Normalizer()
    val v = Vectors.dense(1.0,2.0,3.0)
    //正则化
    println(normalizer.transform(v))
  }
}