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

​

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
本文同样刊载于腾讯云+：https://cloud.tencent.com/developer/article/1031234

注：这篇文章标题包含两个重要的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的主题有Cocktails和Happy hours的相关值，而Happy Hour与bar（饮酒场所），Restaurants以及Cocktails相关。因此，虽然它包含skos，(是)间接将Cocktails与Restaurants连接的相关三元组，但它没有将这些与Space stations有关的主题联系起来，所以Space stations的主题不是与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

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

加载这些数据结构（加上另一个允许快速查找的参考标签）后，我下面的程序将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 ——
")
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。从今天 开始