1. 图论与GraphX
图论是一个数学学科，研究一组实体（称为顶点）之间两两关系（称为边）的特点。通过构建关系图谱，并对关系进行分析，可以实现更好的投放广告，推荐关系等。随着关系图谱越来越强大，计算量也越来越大，于是不断有新的并行图处理框架被开发出来。如谷歌的Pregel、雅虎的 Giraph 和卡内基梅隆大学的 GraphLab。
本章介绍的GraphX是基于Spark 上的一个扩展工具它支持 Pregel、Giraph 和 GraphLab 中的许多图并行处理任务。虽然有些图处理并不如这几个框架快，但是由于它是基于 Spark 的，所以用于数据分析时还是较为方便的。
2. MEDLINE
MEDLINE (Medical Literature Analysis and Retrieval System Online, 医学文献在线分析和检索系统)是一个学术论文数据库，收录发表在生命科学和医学领域期刊上的文献。MEDLINE 引用量非常大而且更新频率快，研究人员在所有文献引用索引上开发了一套全面的语义标签，称为 MeSH（Medical Subject Headings）。这些标签提供了一个有用的框架，使用Mesh，人们就可以在阅读文献时知道文献之间的关系。这里我们会用 Spark 和 GraphX 来获取、转化并分析Mesh 术语网络。
我们的目的是了解文献引用图谱的概况和特点，所以需要从多个角度来研究数据集：
1. 数据集中主要主题和它们的伴生关系
2. 找出数据集中的连通组件（connected component），也就是主题之间是否有连通性
3. 图的度分布：描述了主题的相关度变化，有助于找到与其相关联最多的主题
4. 两个图统计量：聚类系数 和 平均路径长度
3. 获取数据
wget ftp://ftp.nlm.nih.gov/nlmdata/sample/medline/*.gz
解压并检查数据，然后将数据上传到 HDFS:
hdfs dfs -mkdir medline
hdfs dfs -put *.xml medline/
样本中每个条目是一条 MedlineCitation 类型的记录，该记录包含文章在生物医学杂志上的发表信息，包括杂志名称、发行期号、发行日期、作者姓名、摘要、MeSH 关键字集合。此外，MeSH 关键字还有一个属性，用于表示该关键字所指概念是不是文章主要的主题
首先我们需要把 xml 格式的 medline 数据读到 Spark shell 中：
def loadMedline(sc: SparkContext, path: String) = {
@transient val conf = new Configuration()
conf.set(XmlInputFormat.START_TAG_KEY, "<MedlineCitation ")
conf.set(XmlInputFormat.END_TAG_KEY, "</MedlineCitation>")
val in = sc.newAPIHadoopFile(path, classOf[XmlInputFormat], classOf[LongWritable], classOf[Text], conf)
in.map(line => line._2.toString)
}
val medline_raw = loadMedline(sc, "hdfs:///user/hadoop/medline")
这里的 @transient 注释表示某个字段不需要被序列化。newAPIHadoopFile 方法：Get an RDD for a Hadoop file with an arbitrary new API InputFormat。方法定义为：
public <K,V,F extends org.apache.hadoop.mapreduce.InputFormat<K,V>> RDD<scala.Tuple2<K,V>> newAPIHadoopFile(String path, scala.reflect.ClassTag<K> km, scala.reflect.ClassTag<V> vm, scala.reflect.ClassTag<F> fm)
4. 用 scala XML 工具解析XML 文档
import scala.xml._
val raw_xml = medline_raw.take(1)(0)
val elem = XML.loadString(raw_xml)
变量 elem 是 scala.xml.Elem 类的实例，Scala 用 scala.xml.Elem 类表示 XML 文档中的一个节点，该类内置了查询节点信息和节点内容的函数，如：
scala> elem.label
res3: String = MedlineCitation
scala> elem.attributes
res4: scala.xml.MetaData = Status="MEDLINE" Owner="PIP"
它也提供了查找给定 XML 节点子节点的几个运算符，其中第一个就是 ，用于根据名称查询节点的直接节点：
scala> elem "MeshHeadingList"
res5: scala.xml.NodeSeq =
NodeSeq(<MeshHeadingList>
<MeshHeading>
<DescriptorName UI="D001519" MajorTopicYN="N">Behavior</DescriptorName>
</MeshHeading>
<MeshHeading>
<DescriptorName UI="D000013" MajorTopicYN="N">Congenital Abnormalities</DescriptorName>
</MeshHeading>
…
运算符 只对节点的直接节点有效，如果我们运行 elem "MeshHeading" 会得到空的 NodeSeq。为了得到给定节点的间接子节点，我们要用运算符 \：
scala> elem \ "MeshHeading"
res7: scala.xml.NodeSeq =
NodeSeq(<MeshHeading>
<DescriptorName UI="D001519" MajorTopicYN="N">Behavior</DescriptorName>
</MeshHeading>, <MeshHeading>
<DescriptorName UI="D000013" MajorTopicYN="N">Congenital Abnormalities</DescriptorName>
</MeshHeading>, <MeshHeading>
<DescriptorName UI="D006233" MajorTopicYN="N">Disabled Persons</DescriptorName>
…
可以用 \ 运算符直接得到 DescriptorName 条目，并通过在每个 NodeSeq 内部元素上调用 text 函数把每个节点内的 MeSH 标签提取出来：
(elem \ "DescriptorName").map(_.text)
我们在数据里要提取的条目为 MajorTopic 的属性，对此我们可以根据 DescriptorName 条目下的 MajorTopicYN 的值来判断。它表示该 MeSH 标签是否是所引用的文章的主要标题。只要我们在 XML 标签属性前加上“@“符号，就可以用 和 \ 运算符得到 XML 标签属性的值。用此特性我们可以构造一个过滤器，只返回主要属性的 topic：
def majorTopics(elem : Elem): Seq[String] = {
val dn = elem \ "DescriptorName"
val mt = dn.filter( n => (n "@MajorTopicYN").text == "Y")
mt.map(_.text)
}
scala> majorTopics(elem)
res24: Seq[String] = List(Intellectual Disability, Maternal-Fetal Exchange, Pregnancy Complications)
接下来将代码应用到集群：
val melem = medline_raw.map(XML.loadString)
val medline = melem.map(majorTopics)
medline.cache
medline.first
res26: Seq[String] = List(Intellectual Disability, Maternal-Fetal Exchange, Pregnancy Complications)
5. 分析 MeSH 主要主题及其伴生关系
在获取到MeSH 标签后，我们需要知道数据集中标签的总体分布情况，为此我们需要计算一些基本统计量，比如记录条数和主要MeSH 主题出现频率的直方图：
scala> medline.count()
res30: Long = 240000
scala> val topics = medline.flatMap(mesh => mesh)
val topicCount = topics.countByValue()
scala> topicCount.size
res32: Int = 14548
scala> topicCount.toSeq.sortBy(_._2).reverse.take(10).foreach(println)
(Research,1649)
(Disease,1349)
(Neoplasms,1123)
(Tuberculosis,1066)
(Public Policy,816)
(Jurisprudence,796)
(Demography,763)
(Population Dynamics,753)
(Economics,690)
(Medicine,682)
以上结果可以对数据给出一个大致的描述，包括一共有多少个主题，最频繁的主题等。可以看到，我们的数据一共有240000 个文档，最频繁出现的 topic（Research）只占了很少一部分（1649/240000 = 0.06%）。对此，我们猜测包含某个主题的文档的个数的总体分布可能为长尾形态。为了验证此猜想，可以使用以下命令查看数据分布
scala> topicCount.groupBy(_._2).mapValues(_.size).toSeq.sorted.take(10).foreach(println)
(1,3106)
(2,1699)
(3,1207)
(4,902)
(5,680)
(6,571)
(7,490)
(8,380)
(9,356)
(10,296)
当然我们主要关注的还是 MeSH 的伴生主题。MEDLINE 数据集中每一项都是一个字符串列表，代表每个引用记录中提及的主题名称。要得到伴生关系，我们要为这些字符串列表生成一个二元组集合。对此我们可以使用 Scala 集合工具包里的 combinations 方法，它返回的是一个 Iterator，因此并不需要把所有组合都放在内存里，如：
val list = List(1, 2, 3)
val combs = list.combinations(2)
combs.foreach(println)
List(1, 2)
List(1, 3)
List(2, 3)
当用这个函数来生产子列表时，我们要注意所有的列表要按照同样的方式进行排序，以便之后使用 Spark 对它们进行汇总。这个因为 combinations 函数返回的列表取决于输入元素的顺序，而元素相同但属虚不同的两个列表是不等的：
val combs = list.reverse.combinations(2)
combs.foreach(println)
List(3, 2)
List(3, 1)
List(2, 1)
List(3,2) == List(2,3)
res50: Boolean = false
所以在为每条引用记录生成二元自列表集合时，调用 combinations 之前要确保列表是排好序的：
val topicPairs = medline.flatMap(t => t.sorted.combinations(2))
val cooccurs = topicPairs.map(p => (p,1)).reduceByKey(_+_)
cooccurs.cache
cooccurs.count
res60: Long = 213745
我们数据中一共有 14548 个主题，总共可能有 14548 × 14548 / 2 = 105822152 个无序的伴生二元组。然而伴生二元组计算结果显示只有 213745 个，只占可能数量的很少一部分。如果我们查看一下数据中最常出现的伴生二元组：
scala> val ord = Ordering.by[(Seq[String], Int), Int](_._2)
scala> cooccurs.top(10)(ord).foreach(println)
(List(Demography, Population Dynamics),288)
(List(Government Regulation, Social Control, Formal),254)
(List(Emigration and Immigration, Population Dynamics),230)
(List(Acquired Immunodeficiency Syndrome, HIV Infections),220)
(List(Antibiotics, Antitubercular, Dermatologic Agents),205)
(List(Analgesia, Anesthesia),183)
(List(Economics, Population Dynamics),181)
(List(Analgesia, Anesthesia and Analgesia),179)
(List(Anesthesia, Anesthesia and Analgesia),177)
(List(Population Dynamics, Population Growth),174)
以上并未提供特别有用的信息，最常见的伴生二元组与最常见的 topic 非常相关。除此之外，也没有提供什么额外的信息
6. 用 GraphX 来建立一个伴生网络
在研究伴生网络时，标准的数据统计工具并不能提供额外的有价值的信息。我们可以了解数据的统计量等，但是无法了解网络中关系的总体结构。
我们真正想要做的是把伴生网络当作网络来分析：把主题当作图的顶点，把连接两个主题的引用记录看成两个相应顶点之间的边。这样我们就可以计算以网络为中心的统计量。这些网络统计量呢个帮助我们理解网络的总体结构并识别出那些有意思的局部离群点，识别出这些离群点之后我们才需要对其做进一步研究。
GraphX 构建与 Spark 之上，它继承了 Spark 在可扩展性方面的所有特性。这就意味着可以利用 GraphX 对规模极其庞大的图进行分析，这些任务可以在做个分布式的机器上并行执行。
GraphX 针对图计算对 RDD 的实现进行了两项特殊的优化：
1. VertexRDD[VD] 是 RDD[(VertexId, VD)] 的特殊实现，其中 VertexID 类型是 Long 的实例，对每个顶点都是必须的。VD 是顶点关联的任何类型数据，称为顶点属性（vertex attribute）。
2. EdgeRDD[ED] 是 RDD[Edge[ED]] 的特殊实现，其中 Edge 是包含两个 VertexId 值和一个 ED 类型的边属性（edge attribute）。VertexRDD 和 EdgeRDD 在每个数据分区内部均有用于加快连接和属性更新的索引结构。给定 VertexRDD 及其相应的 EdgeRDD 后，我们就能建立一个Graph类的实例，Graph 类包含了许多图计算的高校方法。
要建立一个图，首先要取得用作图顶点标识符的 Long 型值。由于所有主题都是用字符串标识的，因此我们在创建伴生网络时需要将每个主题字符串转换为 64 位的 Long 型值，而且这种方式最好能够分布式执行。
方法之一就是内置的 hashCode 来对任意 Scala 对象产生一个 32 位整数。就这个例子而言，图只有 14548 个顶点，是行得通的。但是如果是对于数百万甚至数千万个顶点的图来说，发生哈希冲突的概率会较高。因此我们选用谷歌开发的 Guava 库中的 Hashing 工具。利用此工具，可以通过 MD5 哈希算法为每个主题生成一个 64 位的唯一标识符：
import com.google.common.hash.Hashing
def hashId(str: String) = {
Hashing.md5().hashString(str).asLong()
}
应用到集群：
val vertices = topics.map(topic => (hashId(topic), topic))
同时，我们需要检查一下是否有哈希冲突：
val uniqueHashes = vertices.map(_._1).countByValue()
val uniqueTopics = vertices.map(_._2).countByValue()
uniqueHashes.size == uniqueTopics.size
res2: Boolean = true
从结果里看到并没有哈希冲突
接下来我们要用前一节中得到的伴生频率计数来生成图的边，方法是使用hash 函数将每个主题映射到相应的顶点 ID。在生成边的时候一个好习惯就是好习惯就是保证左边的 VertexId（GraphX 称为 src）要比右边的 VertexId（GraphX 称为 dst）小。虽然 GraphX 工具包中大多数算法都不要求 src 和 dst 之间有大小关系，但确实有几个算法存在这样的要求。因此最好在一开始就保证大小顺序。
import org.apache.spark.graphx._
val edges = cooccurs.map( p => {
val (topics, cnt) = p
val ids = topics.map(hashId).sorted
Edge(ids(0), ids(1), cnt)
})
把顶点和边都创建好后，就可以创建 Graph 实例了。我们需要将 Graph 缓存起来，这样便于后续处理时使用：
val topicGraph = Graph(vertices, edges)
topicGraph.cache
上面的 vertices 和 edges 参数都是普通 RDD ，但是Graph API 会自动将输入的 RDD 转换为 VertexRDD 和 EdgeRDD，这样顶点计数也不会将重复的顶点算在计数内：
scala> vertices
val vertices: org.apache.spark.rdd.RDD[(Long, String)]
scala> vertices.count
res0: Long = 280464
scala> topicGraph.vertices
val vertices: org.apache.spark.graphx.VertexRDD[String]
scala> topicGraph.vertices.count
res3: Long = 14548
如果某两个顶点二元组在 EdgeRDD 中重复出现，Graph API 不会对其进行去重处理，这样 GraphX 就可以创建多图（multigraph），也就是相同顶点之间可以用多条不同值的边。如果图顶点代表了我许多丰富含义的对象，多图往往是很有用的。多图也可以让我们根据实际情况使用有向边或无向边。
7. 理解网络结构
研究图时，我们需要计算出一些概要统计量，大概了解数据的结构。Graph 类内置了计算一些概要统计量的 API，结合其他常规 Spark RDD API，我们可以很轻松地了解到图的结构。
1. 连通组件
最基本的属性之一就是是否是连通图。如果图是非联通的，那么我们可以将图划分成一组更小的子图，这样就可以分别对每个子图进行研究。
连通性是图的基本属性，可以通过调用 GraphX 的connectedComponents 方法获取：
scala> val connectedComponentGraph = topicGraph.connectedComponents()
connectedComponentGraph: org.apache.spark.graphx.Graph[org.apache.spark.graphx.VertexId,Int] = org.apache.spark.graphx.impl.GraphImpl@67ca9edb
请注意 connectedComponents 方法的返回值，也是一个 Graph 对象，顶点属性的类型是 VertexId，第二个 Int 类型的值表示每个顶点所属连通组件的唯一标识符。想得到连通组件的个数和大小，我们可以在VertexRDD 中的每个顶点的 VertexId 上调用 countByValue 这个方法：
def sortedConnectedComponents(
connectedComponents: Graph[VertexId, _]): Seq[(VertexId, Long)] = {
val componentCounts = connectedComponents.vertices.map(_._2).countByValue
componentCounts.toSeq.sortBy(_._2).reverse
}
val componentCounts = sortedConnectedComponents(connectedComponentGraph)
我们可以看看一共有多少个连通组件，以及前10个最大的连通组件：
scala> componentCounts.size
res7: Int = 878
scala> componentCounts.take(10).foreach(println)
(-9222594773437155629,13610)
(-6100368176168802285,5)
(-1043572360995334911,4)
(-8641732605581146616,3)
(-8082131391550700575,3)
(-5453294881507568143,3)
(-6561074051356379043,3)
(-2349070454956926968,3)
(-8186497770675508345,3)
(-858008184178714577,2)
可以看到最大的连通图包含了超过 90% 的顶点，第二大的连通组件只有5 个顶点。为了弄清楚为什么这些小的连通组件没有和最大的连通组件连通，我们需要看一下它们的主题。为了查看小组件相关的主题名称，我们需要将连通组件图对应的 VertexRDD 和原始概念图执行 join 操作。
VertexRDD 提供了 innerJoin 转换，它利用了 GraphX 的内部数据结构，性能比常规Spark 的join转换要好得多。innerJoin 方法需要我们提供一个函数，该函数的输入为 VertexID 和两个 VertexRDD的内部数据，函数的返回值是一个新的VertexRDD，它是innerJoin 方法结果。对应到我们这里的情况，我们想要知道每个连通组件的概念的名称，因此需要返回一个包含概念名称和组件ID 的二元组：
val nameCID = topicGraph.vertices.innerJoin(connectedComponentGraph.vertices){
(topicId, name, componentId) => (name, componentId)
}
现在我们查看一下第二大连通组件的主题名称：
> val c1 = nameCID.filter( x => x._2._2 == componentCounts(1)._1)
> c1.collect.foreach(x => println(x._2._1))
Acetyl-CoA C-Acyltransferase
Racemases and Epimerases
Enoyl-CoA Hydratase
3-Hydroxyacyl CoA Dehydrogenases
Carbon-Carbon Double Bond Isomerases
在找连通组件时，底层使用了以下方式：
- connectedComponents 方法利用 VertexId 作为顶点唯一标识符在图上执行一些列迭代计算
- 在计算的每个阶段，每个顶点把它所收到的最小VertexID广播到相邻节点。
- 第一次迭代时，这个最小VertexID就是顶点自身的ID，在随后的迭代中该最小VertexID通常会被更新掉。
- 每个顶点都记录它所收到的VertexID的最小值，如果在某一次迭代中，所有顶点的最小VertexID都没有变化，那么连通组件的计算就完成了，每个顶点都将分配给该顶点的最小VertexID所代表的组件
2. 度的分布
为了更多了解图的结构信息，我们需要知道每个顶点的度，也就是每个顶点所属边的条数。对于一个五环图，因为每条边都包含两个不同的顶点，所以全体顶点的度之和等于边的条数的两倍。
GraphX 中我们可以通过在Graph 对象上调用degrees方法得到每个顶点的度。degrees 方法返回一个整数的VertexRDD，其中每个整数代表一个顶点的度。现在我们计算一下图的度：
> val degrees = topicGraph.degrees.cache()
> degrees.map(_._2).stats
res26: org.apache.spark.util.StatCounter = (count: 13721, mean: 31.155892, stdev: 65.497591, max: 2596.000000, min: 1.000000)
可以看到degree RDD 中条目的个数（13721）比图里的顶点数（14548）要少，这是由于有顶点并没有连接边。这可能是由于MEDLINE 数据中某些引用只有一个主要主题词，因此有些主题并未与其他主题同时出现。我们可以通过检查原始RDD medline 来确认我们的推测：
> val sing = medline.filter( _.size == 1)
> sing.count
res31: Long = 44509
> val singTopic = sing.flatMap(topic => topic).distinct()
> singTopic.count
res34: Long = 8243
其中有 8243 个不同主题词单独出现在 MEDLINE 数据库中的 44509 篇文章中。现在我们将已经出现在 topicPairs RDD 出现过的这些主题词去掉：
> val topic2 = topicPairs.flatMap(p => p)
> singTopic.subtract(topic2).count
res39: Long = 827
这会过滤掉 MEDLINE 数据库文档中单独出现的 827 个主题词。14548 - 827 = 13721，正好是 degrees 中的条目数。
虽然此图中度的均值较小，意味着普通顶点只连接到少数几个其他节点，但是度的最大值却表明至少有一个节点是高度连接的，它几乎和图中五分之一的点均有连接。
我们进一步看看那些度很高的点所对应的概念，这里我们仍有 innerJoin 方法，此方法会返回在两个 VertexRDD 中均出现的顶点，因此那些没有与其他概念同时出现的概念将被过滤掉：
def topNamesAndDegrees(degrees: VertexRDD[Int], topicGraph: Graph[String, Int]): Array[(String, Int)] = {
val namesAndDegrees = degrees.innerJoin(topicGraph.vertices) {
(topicId, degree, name) => (name, degree)
}
val ord = Ordering.by[(String, Int), Int](_._2)
namesAndDegrees.map(_._2).top(10)(ord)
}
然后可以查看到度数较高的主题：
scala> topNamesAndDegrees(degrees, topicGraph).foreach(println)
(Research,2596)
(Disease,1746)
(Neoplasms,1202)
(Blood,914)
(Pharmacology,882)
(Tuberculosis,815)
(Toxicology,694)
(Drug Therapy,678)
(Jurisprudence,661)
(Biomedical Research,633)