在美国国会图书馆标题表的SKOS上运行Apache Spark GraphX算法

Running Apache Spark GraphX algorithms on Library of Congress subject heading SKOS

这是Bob DuCharme的一篇客串文章。

原文出现在://www.snee.com/bobdc.blog/2015/04/running-spark-graphx-algorithm.html

译者微博:@从流域到海域

译者博客:blog.csdn.net/solo95

注:这篇文章标题包含两个重要的item,下面是译者翻译的时候的参考,如果你对美国国会图书馆标题表或者 SKOS有疑问,请参考下面的网页:

Library of Congress subject heading(LCSH),是美国国家图书馆自1986年开始维护的对馆藏内容进行分类的系统,可以翻译成国家图书馆标题表。

SKOS (Simple Knowledge Organization System),简单知识组织系统,为语义Web环境下的数字信息资源整合提供了描述和转化机制,解决了信息资源的语义互操作问题。由W3C提出的,在语义网框架下,用机器可理解的语言来表示知识组织系统的一个模型 。

http://id.loc.gov/authorities/subjects.html

http://www.docin.com/p-601048210.html (只需要看第一段)

在美国国会图书馆标题表的SKOS上运行Apache Spark GraphX算法

虽然只是一个算法,但它非常酷。

上个月,在Apache Spark和SPARQL中; RDF Graphs和GraphX(这篇文章中),我描述了Apache Spark如何作为一个更有效地进行MapReduce操作的替代方法出现,以便跨群集分配计算任务。我还描述了Spark的GraphX库如何让您在图形数据结构上进行这种计算,以及我如何获得一些使用RDF数据的想法。我的目标是在GraphX数据上使用RDF技术,或者,以演示(他们彼此)如何互相帮助。我用Scala程序演示了前者,它将一些GraphX数据输出为RDF,然后显示一些在该RDF上运行的SPARQL查询。

(Map/MapReduce操作,参见谷歌一篇非常著名的论文:https://research.google.com/archive/mapreduce.html,它用来进行一系列针对数据集群的大规模数据处理,可以将函数映射到每一个键值对进行处理,直接产生结果键值对。)

今天我将通过读取一个众所周知的RDF数据集并在其上执行GraphX的连接组件算法来演示后者。该算法将节点收集到彼此连接但不连接到其他任何节点的分组中。在经典的大数据场景中,这可以帮助应用程序执行任务,例如识别在大型网络中的子网络,只需要给出基于他们朋友的喜好而推荐的产品或者吸猫视频(cat video)作为线索。

自1898年以来,美国国会图书馆一直在研究其Subject Headings元数据,这些数据在SKOS RDF中也可用。许多主题包括“相关”的值; 例如,您可以看到Cocktails的主题有Cocktail partiesHappy hours的相关值,而Happy Hour与Bars (Drinking establishments)Restaurants,以及Cocktails相关。因此,虽然它包含skos,(是)间接将Cocktails与Reastaurant连接的相关三元组,但它没有将这些与Space stations有关的主题联系起来,所以Space station的主题不是与Cocktails主题相同的Connected Components子句的一部分。

在将美国国会图书馆标题表的RDF(文件)读入GraphX图表并在skos上运行连接组件(Connected Components)算法之后,下面是我在输出开头发现的一些分组:

"Hiding places" 
"Secrecy" 
"Loneliness" 
"Solitude" 
"Privacy" 
--------------------------
"Cocktails" 
"Bars (Drinking establishments)" 
"Cocktail parties" 
"Restaurants" 
"Happy hours" 
--------------------------
"Space stations" 
"Space colonies" 
"Large space structures (Astronautics)" 
"Extraterrestrial bases" 
--------------------------
"Inanna (Sumerian deity)" 
"Ishtar (Assyro-Babylonian deity)" 
"Astarte (Phoenician deity)" 
--------------------------
"Cross-cultural orientation" 
"Cultural competence" 
"Multilingual communication" 
"Intercultural communication" 
"Technical assistance--Anthropological aspects" 
--------------------------

(您可以在这里找到完整的输出(文件),是一个565K的文件)。使用基于RDF的应用程序的人已经知道这种数据可以帮助增强搜索。例如,搜索与“空间站”相关媒体的人可能也会对“太空殖民地”和“外星基地”下的媒体感兴趣。这些数据也可以帮助其他应用程序,现在它可以帮助使用Spark的分布式应用程序。

在GraphX数据结构中存储RDF

首先,正如我在前面的博客中提到的,GraphX开发目前意味着使用Scala编程语言进行代码编写,所以我一直在学习Scala。我的XML老朋友Tony Coates编写了用于处理RDF的一个Scala API,它比我以前的方法能更好地利用本地Scala数据结构,而且banana-rdf Scala library 也看起来很有趣,但尽管我也使用Scala,但我的主要关注点是在Spark GraphX数据结构中存储RDF,特别是在Scala中。

基本的Spark数据结构是弹性分布式数据集(Resilient Distributed Dataset, or RDD)。GraphX使用的图形数据结构是顶点RDD和边RDD的组合。每个RDD都可以有额外的信息; Spark网站的 "Example Property Graph"包含带有顶点的(名称,角色)对和边的描述性属性字符串。在GraphX图中存储RDF的第一步显然是将谓词存储在边RDD,并将顶点RDD中的主体和资源对象以及文字属性作为这些RDD中的额外信息,如(名称,角色)对和Spark网站的Example Property Graph中描述边的字符串。

((名称,角色)对是2个元素组成的一个数据。)

但是,正如我上次写的那样,一个使用用硬编码RDF的人会问这些问题

  • 那边的属性呢?例如,要是我想说一个"xp:advisor"是一个rdfs:subPropertyOf the Dublin Core property dc:rdfs:subPropertyOf the Dublin Core property dc:呢?
  • 将属性(如“rxin”的名称和“student”的角色)分配给像3L节点的这个功能是很好的,但是如果我没有一套一致的属性分配给每个节点呢?,比如,如果我汇总了两个来自不同来源的不使用所有相同属性来描述的人员数据这些人员的。

The Example Property Graph可以将这些(名称,角色)对与顶点存储在一起,因为RDD被声明为RDD(VertexId,(String, String))。每个顶点将有两个字符串存储在一起; 不多也不少。这是一个数据结构,但是你也可以把它看作是一个规范的模式,上面的第二个问题是问如何解决这个问题。

我通过将数据存储在三个数据结构(上述两个RDD和另外一个RDD)中来解决了这两个问题:

  • 对于顶点RDD,以及必须存储为每个顶点标识符所需的长整数,我只存储了一个额外的信息:与该RDF资源相关联的URI。我为主语做了这些,谓词(它可能不是GraphX意义上的“顶点”,但是该死的,如果可以的话我希望它们是资源,是三元组的主语或宾语),以及相关对象。在读取了{ http://id.loc.gov/authorities/subjects/sh85027617 http://www.w3.org/2004/02/skos/core#related <http:// id。 loc.gov/authorities/subjects/sh2009010761> }美国国会图书馆的数据的三元数据后,该项目将在这个RDD上建立节点标识符可能是1L,2L,3L的三个顶点,将每一个三重的URI与其中的RDD顶点存储在一起。
  • 对于边RDD,连同需要的两个长整数来标识边的开始和结束的顶点,每个边也存储相关谓词的URI作为边的“描述”。以上三项的边将是(1L,3L,http://www.w3.org/2004/02/skos/core#related)。
  • 为了增加从上面两个RDD创建的图数据结构,我创建了第三个RDD来存储文字属性值。每个条目存储表示具有该属性的资源的顶点的长整数,一个表示属性(在顶点RDD中分配给该属性的整数)的长整数,以及表示属性值的字符串。对于三元组{ http://id.loc.gov/authorities/subjects/sh2009010761 http://www.w3.org/2004/02/skos/core#prefLabel“Happy hours”}假设已经存储了4L作为skos:prefLabel属性的内部标识符,它可能会存储(3L,4L,“Happy hours”)。为了运行连接组件算法(Connected Components algorithm ),然后输出每个子图的每个成员的参考标签,我不需要这个RDD,但是它为在Spark GraphX程序中使用RDF来做什么开辟了许多可能性。

创建一个国会图书馆标题表连接组件的报告

加载这些数据结构(加上另一个允许快速查找的参考标签)后,我下面的程序将GraphX连接组件算法应用到使用skos:related属性连接顶点的图的子集,如“Cocktails”和“Happy hours”。遍历结果时,它使用它们加载一个哈希映射,这个映射带有连接组件的每个子图的列表。然后,它会遍历每个列表,打印与每个子图的每个成员关联的标签以及一串连字符(即"-"),以显示每个列表的结束位置,如上面的摘录所示。

我不会更详细地介绍我的程序中的内容,因为我非常重视它。(我不得不感谢上面提到的我的朋友Tony,因为他之前帮助我走出了被Scala范围问题困扰的一个节点,而且,正如我之前提醒的那样,我的编码风格可能会让有经验的Scala程序员喝红牛被呛到,我也很乐意听取有关改进的建议。)

在让程序正常运行一小部分数据之后,我把它运行在我从国会图书馆下载的有7,705,147三元组的1 GB的" subject-skos-2014-0306.nt"文件上。Spark通过给你一个基础架构来分配以执行跨多台机器的程序,从而使得应用程序可以扩展,但是我的单机上的8GB还不足以运行这个,所以我使用了两个grep命令来创建一个只有skos:related和skos:prefLabel的三元组。在此时,我总共有439,430个三元组。由于我的代码没有考虑到空白节点,我删除了使用它们(空白结点)的385个三元组,剩下439045个(三元组)在60MB文件中。这个可以成功运行,您可以按照前面显示的链接查看完整的输出。

其他在您的RDF数据上运行GraphX算法

除连接组件(Connected Components)之外的其他GraphX算法有Page Rank和Triangle Counting。图论是一个有趣的世界,其中我最喜欢的一个词组是“ Strangulated graph ”。

(Strangulated graph指一个所有环都是三角形的无向图,参见维基百科,有译为绞窄性图的但无法佐证,译者注)

关于RDF和数据关联技术( Linked Data technology)的最大的一件事情就是越来越多的有趣数据被公开发布,而且可以使用这些算法作为新工具进而使用这些数据进行工作,这些工具可以在比典型的Hadoop MapReduce jobs更便宜,更快进行扩展的集群上运行 - (这里)有很多很大的可能性。

//////////////////////////////////////////////////////////////////
// readLoCSH.scala: read Library of Congress Subject Headings into
// Spark GraphX graph and apply connectedComponents algorithm to those
// connected by skos:related property.

import scala.io.Source
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.graphx._
import org.apache.spark.rdd.RDD
import scala.collection.mutable.ListBuffer
import scala.collection.mutable.HashMap

object readLoCSH {
val componentLists = HashMap[VertexId, ListBuffer[VertexId]]()
val prefLabelMap =  HashMap[VertexId, String]()
def main(args: Array[String]) {
val sc = new SparkContext("local", "readLoCSH", "127.0.0.1")

// regex pattern for end of triple
val tripleEndingPattern = """\s*\.\s*$""".r
// regex pattern for language tag
val languageTagPattern = "@[\\w-]+".r    

// Parameters of GraphX Edge are subject, object, and predicate
// identifiers. RDF traditionally does (s, p, o) order but in GraphX
// it’s (edge start node, edge end node, edge description).
// Scala beginner hack: I couldn’t figure out how to declare an empty
// array of Edges and then append Edges to it (or how to declare it
// as a mutable ArrayBuffer, which would have been even better), but I
// can append to an array started like the following, and will remove
// the first Edge when creating the RDD.
var edgeArray = Array(Edge(0L,0L,"http://dummy/URI"))
var literalPropsTriplesArray = new Array[(Long,Long,String)](0)
var vertexArray = new Array[(Long,String)](0)

// Read the Library of Congress n-triples file
//val source = Source.fromFile("sampleSubjects.nt","UTF-8")  // shorter for testing
val source = Source.fromFile("PrefLabelAndRelatedMinusBlankNodes.nt","UTF-8")
val lines = source.getLines.toArray

// When parsing the data we read, use this map to check whether each
// URI has come up before.
var vertexURIMap = new HashMap[String, Long];

// Parse the data into triples.
var triple = new Array[String](3)
var nextVertexNum = 0L

for (i <- 0 until lines.length) {
    // Space in next line needed for line after that.
    lines(i) = tripleEndingPattern.replaceFirstIn(lines(i)," ")
    triple = lines(i).mkString.split(">\\s+")       // split on "> "

    // Variables have the word "triple" in them because "object"
    // by itself is a Scala keyword.

    val tripleSubject = triple(0).substring(1)   // substring() call
    val triplePredicate = triple(1).substring(1) // to remove "<"

    if (!(vertexURIMap.contains(tripleSubject))) {
        vertexURIMap(tripleSubject) = nextVertexNum
        nextVertexNum += 1

    }

    if (!(vertexURIMap.contains(triplePredicate))) {
        vertexURIMap(triplePredicate) = nextVertexNum
        nextVertexNum += 1

    }

    val subjectVertexNumber = vertexURIMap(tripleSubject)
    val predicateVertexNumber = vertexURIMap(triplePredicate)

    // If the first character of the third part is a <, it’s a URI;
    // otherwise, a literal value. (Needs more code to account for
    // blank nodes.)

    if (triple(2)(0) == ‘<‘) {
        val tripleObject = triple(2).substring(1)   // Lose that <.
        if (!(vertexURIMap.contains(tripleObject))) {
            vertexURIMap(tripleObject) = nextVertexNum
            nextVertexNum += 1
        }
        val objectVertexNumber = vertexURIMap(tripleObject)
        edgeArray = edgeArray :+
            Edge(subjectVertexNumber,objectVertexNumber,triplePredicate)
    }
    else {
        literalPropsTriplesArray = literalPropsTriplesArray :+
            (subjectVertexNumber,predicateVertexNumber,triple(2))
    }
}

// Switch value and key for vertexArray that we’ll use to create the
// GraphX graph.
for ((k, v) <- vertexURIMap) vertexArray = vertexArray :+  (v, k)   

// We’ll be looking up a lot of prefLabels, so create a hashmap for them.
for (i <- 0 until literalPropsTriplesArray.length) {
    if (literalPropsTriplesArray(i)._2 ==
        vertexURIMap("http://www.w3.org/2004/02/skos/core#prefLabel")) {
        // Lose the language tag.
        val prefLabel =
            languageTagPattern.replaceFirstIn(literalPropsTriplesArray(i)._3,"")
        prefLabelMap(literalPropsTriplesArray(i)._1) = prefLabel;
    }
}

// Create RDDs and Graph from the parsed data.
// vertexRDD Long: the GraphX longint identifier. String: the URI.
val vertexRDD: RDD[(Long, String)] = sc.parallelize(vertexArray)

// edgeRDD String: the URI of the triple predicate. Trimming off the
// first Edge in the array because it was only used to initialize it.
val edgeRDD: RDD[Edge[(String)]] = sc.parallelize(edgeArray.slice(1,edgeArray.length))

// literalPropsTriples Long, Long, and String: the subject and predicate
// vertex numbers and the the literal value that the predicate is
// associating with the subject.
val literalPropsTriplesRDD: RDD[(Long,Long,String)] = sc.parallelize(literalPropsTriplesArray)
val graph: Graph[String, String] = Graph(vertexRDD, edgeRDD)

// Create a subgraph based on the vertices connected by SKOS "related"

// property.
val skosRelatedSubgraph = graph.subgraph(t => t.attr =="http://www.w3.org/2004/02/skos/core#related")

// Find connected components  of skosRelatedSubgraph.
val ccGraph = skosRelatedSubgraph.connectedComponents() 

// Fill the componentLists hashmap.
skosRelatedSubgraph.vertices.leftJoin(ccGraph.vertices) {
case (id, u, comp) => comp.get
}.foreach
{ case (id, startingNode) =>
  {
      // Add id to the list of components with a key of comp.get
      if (!(componentLists.contains(startingNode))) {
          componentLists(startingNode) = new ListBuffer[VertexId]
      }
      componentLists(startingNode) += id
  }
}
// Output a report on the connected components.
println("——  connected components in SKOS \"related\" triples ——\n")
for ((component, componentList) <- componentLists){
    if (componentList.size > 1) { // don’t bother with lists of only 1
        for(c <- componentList) {
            println(prefLabelMap(c));
        }
        println("————————–")
    }
}
sc.stop
}

免费试用Databricks。从今天 开始

本文的版权归 Steve Wang 所有,如需转载请联系作者。

编辑于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏沈唁志

如何优化PHP性能呢?PHP性能优化总结

1493
来自专栏大数据

No.67 Hadoop 实践案例——记录去重

转载声明 本文为灯塔大数据原创内容,欢迎个人转载至朋友圈,其他机构转载请在文章开头标注:转自:灯塔大数据;微信:DTbigdata 编者按:灯塔大数据将每周持续...

1828
来自专栏小灰灰

时序数据库InfluxDB基本概念小结

InfluxDB作为时序数据库,与传统的关系型数据库相比而言,还是有一些区别的,下面尽量以简单明了的方式介绍下相关的术语概念

1454
来自专栏雪胖纸的玩蛇日常

第一章 介绍与循环

1363
来自专栏mySoul

mongodb

NoSQL不使用SQL作为查询语言。其数据的储存可以不需要固定的表格形式。也会经常的被使用sql的join

1110
来自专栏CDA数据分析师

如何在R中操作非结构化数据?

本文由CDA作者库成员HarryZhu原创,并授权发布。 CDA作者库凝聚原创力量,只做更有价值的分享。 ? 介绍 现代化数据科学中的 DataFrame 概念...

1999
来自专栏瓜大三哥

读懂用好Timing Report

静态时序分析 静态时序分析(Static Timing Analysis)简称STA,采用穷尽的分析方法来提取出整个电路存在的所有时序路径,计算信号在这些路径...

2385
来自专栏AILearning

Apache Spark 2.2.0 中文文档 - SparkR (R on Spark) | ApacheCN

SparkR (R on Spark) 概述 SparkDataFrame 启动: SparkSession 从 RStudio 来启动 创建 S...

3815
来自专栏Java架构师历程

solr

Solr它是一种开放源码的、基于 Lucene Java 的搜索服务器,易于加入到 Web 应用程序中。Solr 提供了层面搜索(就是统计)、命中醒目显示并且支...

882
来自专栏文渊之博

pyspark 内容介绍(一)

pyspark 包介绍 子包 pyspark.sql module pyspark.streaming module pyspark.ml package py...

3806

扫码关注云+社区