Spark图处理GraphX学习笔记!-创新互联-成都创新互联网站建设

关于创新互联

多方位宣传企业产品与服务 突出企业形象

公司简介 公司的服务 荣誉资质 新闻动态 联系我们

Spark图处理GraphX学习笔记!-创新互联

Spark图处理GraphX学习笔记!

创新互联主要从事成都网站设计、成都网站制作、网页设计、企业做网站、公司建网站等业务。立足成都服务全州,10多年网站建设经验,价格优惠、服务专业,欢迎来电咨询建站服务:028-86922220

一、什么是GraphX?

Graphx利用了Spark这样了一个并行处理框架来实现了图上的一些可并行化执行的算法。

  • 算法是否能够并行化与Spark本身无关

  • 算法并行化与否的本身,需要通过数学来证明

  • 已经证明的可并行化算法,利用Spark来实现会是一个错的选择,因为Graphx支持pregel的图计算模型

二、Graphx包含哪些组件和基本框架?

1、成员变量

graph中重要的成员变量分别为

  1. vertices

  2. edges

  3. triplets

为什么要引入triplets呢,主要是和Pregel这个计算模型相关,在triplets中,同时记录着edge和vertex. 具体代码就不罗列了。

2、成员函数

函数分成几大类

  1. 对所有顶点或边的操作,但不改变图结构本身,如mapEdges, mapVertices

  2. 子图,类似于集合操作中的filter subGraph

  3. 图的分割,即paritition操作,这个对于Spark计算来说,很关键,正是因为有了不同的Partition,才有了并行处理的可能, 不同的PartitionStrategy,其收益不同。最容易想到的就是利用Hash来将整个图分成多个区域。

  4. outerJoinVertices 顶点的外连接操作

三、图的运算和操作 GraphOps

图的常用算法是集中抽象到GraphOps这个类中,在Graph里作了隐式转换,将Graph转换为GraphOps,具体有下列12个算子:

  1. collectNeighborIds

  2. collectNeighbors

  3. collectEdges

  4. joinVertices

  5. filter

  6. pickRandomVertex

  7. pregel

  8. pageRank

  9. staticPageRank

  10. connectedComponents

  11. triangleCount

  12. stronglyConnectedComponents

RDD

RDD是Spark体系的核心,那么Graphx中引入了哪些新的RDD呢,有俩,分别为

  1. VertexRDD

  2. EdgeRDD

较之EdgeRdd,VertexRDD更为重要,其上的操作也很多,主要集中于Vertex之上属性的合并,说到合并就不得不扯到关系代数和集合论,所以在VertexRdd中能看到许多类似于sql中的术语,如

  • leftJoin

  • innerJoin

四、GraphX场景分析

1、图的存储和加载

在进行数学计算的时候,图用线性代数中的矩阵来表示,那么如何进行存储呢?

学数据结构的时候,老师肯定说过好多的办法,不再啰嗦了。

不过在大数据的环境下,如果图很巨大,表示顶点和边的数据不足以放在一个文件中怎么办? 用HDFS

加载的时候,一台机器的内存不足以容下怎么办? 延迟加载,在真正需要数据时,将数据分发到不同机器中,采用级联方式。

一般来说,我们会将所有与顶点相关的内容保存在一个文件中vertexFile,所有与边相关的信息保存在另一个文件中edgeFile。

生成某一个具体的图时,用edge就可以表示图中顶点的关联关系,同时图的结构也表示出来了。

下面是Spark官方示例,用2个Array构造了一个Graph。

val users: RDD[(VertexId, (String, String))] =

 sc.parallelize(Array((3L, ("rxin", "student")), (7L, ("jgonzal", "postdoc")),

            (5L, ("franklin", "prof")), (2L, ("istoica", "prof"))))

// Create an RDD for edges

val relationships: RDD[Edge[String]] =

 sc.parallelize(Array(Edge(3L, 7L, "collab"),   Edge(5L, 3L, "advisor"),

            Edge(2L, 5L, "colleague"), Edge(5L, 7L, "pi")))

// Define a default user in case there are relationship with missing user

val defaultUser = ("John Doe", "Missing")

// Build the initial Graph

val graph = Graph(users, relationships, defaultUser)

2、GraphLoader

graphLoader是graphx中专门用于图的加载和生成,最重要的函数就是edgeListFile。

//以顶点划分,分成4个分区

val graph = GraphLoader.edgeListFile(sc,"hdfs://192.168.0.10:9000/input/graph/web-Google.txt",minEdgePartitions = 4)

五、GraphX应用举例

一行代码:

val rank = graph.pageRank(0.01).vertices

用RDD实现:

完整代码

// Connect to the Spark clusterval 
sc = new SparkContext("spark://master.amplab.org", "research")
// Load my user data and parse into tuples of user id and attribute list
val users = (sc.textFile("graphx/data/users.txt")
  .map(line => line.split(","))
  .map( parts => (parts.head.toLong, parts.tail) ))
  // Parse the edge data which is already in userId -> userId format
  val followerGraph = GraphLoader.edgeListFile(sc, "graphx/data/followers.txt")
  // Attach the user attributes
  val graph = followerGraph.outerJoinVertices(users) { 
   case (uid, deg, Some(attrList)) => attrList  
   // Some users may not have attributes so we set them as empty
    case (uid, deg, None) => Array.empty[String]
    }
// Restrict the graph to users with usernames and names
val subgraph = graph.subgraph(vpred = (vid, attr) => attr.size == 2)
// Compute the PageRank

// Get the attributes of the top pagerank users
val userInfoWithPageRank = subgraph.outerJoinVertices(pagerankGraph.vertices) { 
 case (uid, attrList, Some(pr)) => (pr, attrList.toList) 
  case (uid, attrList, None) => (0.0, attrList.toList)
}

println(userInfoWithPageRank.vertices.top(5)(Ordering.by(_._2._1)).mkString("\n"))

另外有需要云服务器可以了解下创新互联scvps.cn,海内外云服务器15元起步,三天无理由+7*72小时售后在线,公司持有idc许可证,提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案,具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势,专为企业上云打造定制,能够满足用户丰富、多元化的应用场景需求。


标题名称:Spark图处理GraphX学习笔记!-创新互联
转载来于:http://kswsj.cn/article/poocs.html

其他资讯