首页 首页 大数据 大数据入门 查看内容

为什么Spark将成为数据科学家的统一平台

木马童年 2018-12-7 19:37 5 0

数据科学是一个广阔的领域。我自认是一个数据科学家,但和另外一批数据科学家又有很多的不同。数据科学家通常分为统计科学家和数据工程师两个阵营,而我正处于第二阵营。统计科学家使用交互式的统计工具(比如R)来回 ...

数据科学是一个广阔的领域。我自认是一个数据科学家,但和另外一批数据科学家又有很多的不同。数据科学家通常分为统计科学家和数据工程师两个阵营,而我正处于第二阵营。

统计科学家使用交互式的统计工具(比如R)来回答数据中的问题,获得全景的认识。与之相比,数据工程师则更像一名程序员,他们在服务器上编写代码,创建和应用机器学习模型,熟悉C++和Java等系统级语言,经常需要和企业级数据中心的某些组件打交道,比如Hadoop。

而有的数据科学家专注于更细的领域,就像精通R但从未听说过Python或者scikit-learn(反之亦然),即便两者都提供了丰富的统计库。

不完美的统计工具

如果可以提供一种统一的工具,运行在统一的架构,用统一的语言编程,并可以同时满足统计科学家和数据工程师的需求,那该多好啊。我一开始就精通Java,难道为了研究数据,我就必须去学一种像Python或R的语言?我一直使用传统的数据分析工具,难道为了应对大规模计算,就必须去懂MapReduce?正是统计工具的不完美造就了这种局面:

R提供了一个丰富的统计分析和机器学习的解释器。但R难以在分布式条件下执行数据的分析和清洗,以便开展其所擅长的数据分析,也不以一种主流的开发语言为人所知。

Python是一种通用的编程语言,也不乏出色的第三方数据分析库(像Pandas和scikit-learn),但Python也有和R一样的缺陷:只能局限在处理单机能负载的数据量。

在经典的MapReduce计算框架上开发分布式的机器学习算法是可行的(参考Mahout),但程序员需要从零开始,更别说移植复杂计算的难度。

为降低复杂计算移植到MapReduce的难度,Crunch提供一个简单的、傻瓜式的Java API,但MapReduce天生决定了它在迭代计算方面是低效的,尽管大多数机器学习算法都需要迭代计算。

其他的数据科学工具一样无法尽善尽美。基于Java和Hadoop的背景,我开始幻想一个理想的数据科学利器:一个像R和Python的能实现RPEL(读取-估值-打印-循环)的自带统计库函数的命令行解释器,又具备天然的分布式可扩展的属性;拥有像Crunch一样的分布式集合,而且能通过命令行解释器调用。

Spark的优势

这就是Spark让我兴奋的原因。大部分人讨论到Spark时,总是注意到将数据驻留内存以提高计算效率的方面(相对MapReduce),但对我来说这根本不是关键。Spark拥有许多的特征,使之真正成为一个融合统计科学和数据工程的交叉点:

Spark附带了一个机器学习库MLib,虽然只是在初始阶段。

Spark是用Scala语言编写的,运行在Java虚拟机上,同时也提供像R和Python的命令行解释器。

对Java程序员,Scala的学习曲线是比较陡峭的,但所幸Scala可以兼容一切的Java库。

Spark的RDD(弹性分布式数据集),是Crunch开发者熟知的一种数据结构。

Spark模仿了Scala的集合计算API,对Java和Scala开发者来说耳熟能详,而Python开发者也不难上手,而Scala对统计计算的支持也不错。

Spark和其底层的Scala语言,并不只是为机器学习而诞生的,除此之外,像数据访问、日志ETL和整合都可以通过API轻松搞定。就像Python,你可以把整个数据计算流程搬到Spark平台上来,而不仅仅是模型拟合和分析。

在命令行解释器中执行的代码,和编译后运行的效果相同。而且,命令行的输入可以得到实时反馈,你将看到数据透明地在集群间传递与计算。

Spark和MLib还有待完善:整个项目有不少bug,效率也还有提升的空间,和YARN的整合也存在问题。Spark还没办法提供像R那样丰富的数据分析函数。但Spark已然是世界上最好的数据平台,足以让来自任何背景的数据科学家侧目。

实战:Stack Overflow问题的自动标注

Stack Overflow是一个著名的软件技术问答平台,在上面提的每个问题有可能被打上若干个短文本的标签,比如java或者sql,我们的目标在于建立一套系统,使用ALS推荐算法,为新问题的标签提供预测和建议。从推荐系统的角度,你可以把问题想象成user,把标签想象成item。

首先,从Stack Overflow下载官方提供的截至20140120的问答数据stackoverflow.com-Posts.7z。

这是一个能够直接用于分布式计算的bzip格式文件,但在我们的场景下,必须先解压并拷贝到HDFS:

bzcat stackoverflow.com-Posts.7z | hdfs dfs -put – /user/srowen/Posts.xml

解压后的文件大约是24.4GB,包含210万个问题,1800万个回答,总共标注了930万个标签,这些标签排重之后大概是34000个。

确认机器安装了Spark之后,输入spark-shell即可打开Scala的REPL环境。首先,我们读取一个存储在HDFS的Posts.xml文件:

val postsXML = sc.textFile(“hdfs:///user/srowen/Posts.xml”)

这时命令行工具会返回:

postsXML: org.apache.spark.rdd.RDD[String] = MappedRDD[1] at textFile at :12

显示文本文件已转化为一个String型的RDD,你可以通过调用RDD的函数,实现任意的查询运算。比如统计文件的行数:

postsXML.count

这条指令生成大量的输出,显示Spark正在利用分布式的环境计数,最终打印出18066983。

下一步,将XML文件的每一行都存入形如(questionID, tag)的元组。得益于Scala的函数式编程的风格,RDD和Scala集合一样可以使用map等方法:

val postIDTags = postsXML.flatMap { line => // Matches Id=”…” … Tags=”…” in line

val idTagRegex = “Id=\”(\\d+)\”.+Tags=\”([^\”]+)\””.r // // Finds tags like value from above

val tagRegex = “<([^&]+)>”.r // Yields 0 or 1 matches:

idTagRegex.findFirstMatchIn(line) match { // No match — not a line

case None => None // Match, and can extract ID and tags from m

case Some(m) => {

val postID = m.group(1).toInt

val tagsString = m.group(2) // Pick out just TAG matching group

val tags = tagRegex.findAllMatchIn(tagsString).map(_.group(1)).toList // Keep only question with at least 4 tags, and map to (post,tag) tuples

if (tags.size >= 4) tags.map((postID,_)) else None

}

} // Because of flatMap, individual lists will concatenate

// into one collection of tuples}

你会发现这条指令的执行是立即返回的,而不像count一样需要等待,因为到目前为止,Spark并未启动任何主机间的数据变换。

ALS的MLib实现必须使用数值ID而非字符串作为惟一标识,而问题的标签数据是字符串格式的,所以需要把字符串哈希成一个非负整数,同时保留非负整数到字符串的映射。这里我们先定义一个哈希函数以便复用。

def nnHash(tag: String) = tag.hashCode & 0x7FFFFFvar tagHashes = postIDTags.map(_._2).distinct.map(tag =>(nnHash(tag),tag))

现在把元组转换为ALS计算所需的输入:

import org.apache.spark.mllib.recommendation._// Convert to Rating(Int,Int,Double) objects

val alsInput = postIDTags.map(t => Rating(t._1, nnHash(t._2), 1.0))// Train model with 40 features, 10 iterations of ALS

val model = ALS.trainImplicit(alsInput, 40, 10)

这一步生成特征矩阵,可以被用来预测问题与标签之间的关联。由于目前MLib还处于不完善的状态,没有提供一个recommend的接口来获取建议的标签,我们可以简单定义一个:

def recommend(questionID: Int, howMany: Int = 5): Array[(String, Double)] = {

// Build list of one question and all items and predict value for all of them

val predictions = model.predict(tagHashes.map(t => (questionID,t._1)))

// Get top howMany recommendations ordered by prediction value

val topN = predictions.top(howMany)(Ordering.by[Rating,Double](_.rating))

// Translate back to tags from IDs topN.map(r => (

tagHashes.lookup(r.product)(0), r.rating))}

通过上述函数,我们可以获得任意一个问题比如ID为7122697的How to make substring-matching query work fast on a large table?的至少4个标签:

recommend(7122697).foreach(println)

推荐结果如下所示:

(sql,0.17745152481166354)

(database,0.13526622226672633)

(oracle,0.1079428707621154)

(ruby-on-rails,0.06067207312463499)

(postgresql,0.050933613169706474)

注意:

– 每次运行得到的结果不尽相同,是因为ALS是从随机解开始迭代的

– 如果你希望获得实时性更高的结果,可以在recommend前输入tagHashes = tagHashes.cache

真实的问题标签是postgresql、query-optimization、substring和text-search。不过,预测结果也有一定的合理性(postgresql经常和ruby-on-rails一起出现)。

当然,以上的示例还不够优雅和高效,但是,我希望所有来自R的分析师、鼓捣Python的黑客和熟悉Hadoop的开发者,都能从中找到你们熟悉的部分,从而找到一条适合你们的路径去探索Spark,并从中获益。

在不久的将来,多智时代一定会彻底走入我们的生活,有兴趣入行未来前沿产业的朋友,可以收藏多智时代,及时获取人工智能、大数据、云计算和物联网的前沿资讯和基础知识,让我们一起携手,引领人工智能的未来!

数据科学 数据科学家 工程师 程序员 机器学习 数据中心
0
为您推荐
腾讯高级数据分析师彭远权:详解腾讯数据挖掘体系及应用

腾讯高级数据分析师彭远权:详解腾讯数据挖

本文主要是腾讯高级数据分析师彭远权详解腾讯数据挖掘体系及应用,28页PPT的阐述了:…...

互联网数据化运营基础应用之信息质量模型

互联网数据化运营基础应用之信息质量模型

信息质量模型在互联网行业和互联网数据化运营中也是有着广泛基础性应用的。具体来说,…...

全球电商数据版图 市场份额一目了然

全球电商数据版图 市场份额一目了然

不要以为只有中国电商处在“战国阶段”,如火如荼。了解其他国家的电商发展情况,即可…...

21个必知数据科学面试题和答案part1(12-21)

21个必知数据科学面试题和答案part1(12-21)

最近KDnuggets上发的“20个问题来分辨真假数据科学家”这篇文章非常热门,获得了一月…...

揭秘:微信支付商户系统的架构

揭秘:微信支付商户系统的架构

作者:李跃森,腾讯云PostgreSQL首席架构师,腾讯数据库团队架构师,PostgreSQL-x2社…...

中美移动互联网用户行为十“同”九“异“

中美移动互联网用户行为十“同”九“异“

《2014年中美移动互联网调查报告》是国家广告研究院互动营销实验室与美国互动广告局(I…...