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

简介: 获取公开数据集UCL机器学习知识库:包括近300个不同大小和类型的数据集,可用于分类、回归、聚类 和推荐系统任务。数据集列表位于:http://archive.ics.uci.edu/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):之后,它通常会删除常见的单词,比如theandbut
(这些词被称作停用词)。


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

  向量化(vectorization):最后一步就是用向量来表示处理好的词。二元向量可能是最为 简单的表示方式。它用10来分别表示是否存在某个词。从根本上说,这与之前提到的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))
  }
}
目录
相关文章
|
2月前
|
分布式计算 关系型数据库 MySQL
大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL
大数据-88 Spark 集群 案例学习 Spark Scala 案例 SuperWordCount 计算结果数据写入MySQL
52 3
|
4月前
|
存储 分布式计算 Java
|
4月前
|
分布式计算 监控 大数据
如何处理 Spark 中的倾斜数据?
【8月更文挑战第13天】
259 4
|
4月前
|
存储 缓存 分布式计算
|
4月前
|
SQL 存储 分布式计算
|
4月前
|
分布式计算 Apache 数据安全/隐私保护
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
流计算引擎数据问题之在 Spark Structured Streaming 中水印计算和使用如何解决
56 1
|
5月前
|
分布式计算 数据处理 流计算
实时计算 Flink版产品使用问题之使用Spark ThriftServer查询同步到Hudi的数据时,如何实时查看数据变化
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
6月前
|
弹性计算 分布式计算 DataWorks
DataWorks产品使用合集之spark任务如何跨空间取表数据
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
43 1
|
6月前
|
SQL 分布式计算 HIVE
实时计算 Flink版产品使用问题之同步到Hudi的数据是否可以被Hive或Spark直接读取
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
6月前
|
SQL 分布式计算 NoSQL
使用Spark高效将数据从Hive写入Redis (功能最全)
使用Spark高效将数据从Hive写入Redis (功能最全)
402 1