基于pyspark GraphFrames实现图查询和计算

朱雀 2023-07-15 15:16 62阅读 0赞

### 基于pyspark GraphFrames实现图查询和计算 ###

*  GraphFrames基本操作
 *   *  创建图
     *  展示顶点和边的数据
     *  统计顶点的入度和出度
     *  对顶点和边的数据进行分析
     *  搜索指定结构路径
     *  对搜索结果过滤
     *  多路径条件搜索
     *  匿名顶点和边
     *  设置路径不存在的条件
 *  有状态和无状态查询
 *   *  无状态查询
     *  有状态查询
 *  子图
 *   *  例一
     *  例二
 *  参考

# GraphFrames基本操作 #

GraphFrames，该类库是构建在Spark DataFrames之上，它既能利用DataFrame良好的扩展性和强大的性能，同时也为Scala、Java和Python提供了统一的图处理API。GraphX基于RDD API，不支持Python API； 但GraphFrame基于DataFrame，并且支持Python API。

## 创建图 ##

创建图的方式很简单，分别向GraphFrame中传入一个顶点数据集和一个边数据集即可。

from pyspark import SparkContext
    from pyspark.sql import SQLContext
    from graphframes import GraphFrame
    
    
    sc = SparkContext("local", appName="mysqltest")
    sqlContext = SQLContext(sc)
    
    vertices = sqlContext.createDataFrame([
      ("a", "Alice", 34),
      ("b", "Bob", 36),
      ("c", "Charlie", 30),
      ("d", "David", 29),
      ("e", "Esther", 32),
      ("f", "Fanny", 36),
      ("g", "Gabby", 60)], ["id", "name", "age"])
    
    edges = sqlContext.createDataFrame([
      ("a", "b", "friend"),
      ("b", "c", "follow"),
      ("c", "b", "follow"),
      ("f", "c", "follow"),
      ("e", "f", "follow"),
      ("e", "d", "friend"),
      ("d", "a", "friend"),
      ("a", "e", "friend")
    ], ["src", "dst", "relationship"])
    
    # 生成图
    g = GraphFrame(vertices, edges)
    print(g)
    # GraphFrame(v:[id: string, name: string ... 1 more field], e:[src: string, dst: string ... 1 more field])
    print(type(g))
    # <class 'graphframes.graphframe.GraphFrame'>

## 展示顶点和边的数据 ##

# 展示顶点和边的数据
    g.vertices.show()
    # +---+-------+---+
    # | id|   name|age|
    # +---+-------+---+
    # |  a|  Alice| 34|
    # |  b|    Bob| 36|
    # |  c|Charlie| 30|
    # |  d|  David| 29|
    # |  e| Esther| 32|
    # |  f|  Fanny| 36|
    # |  g|  Gabby| 60|
    # +---+-------+---+
    
    g.edges.show()
    # +---+---+------------+
    # |src|dst|relationship|
    # +---+---+------------+
    # |  a|  b|      friend|
    # |  b|  c|      follow|
    # |  c|  b|      follow|
    # |  f|  c|      follow|
    # |  e|  f|      follow|
    # |  e|  d|      friend|
    # |  d|  a|      friend|
    # |  a|  e|      friend|
    # +---+---+------------+

## 统计顶点的入度和出度 ##

入度和出度就是顶点被多少条边所引用，还有边是指向顶点还是由顶点指向其他点。

g.inDegrees.sort("id").show()
    # +---+--------+
    # | id|inDegree|
    # +---+--------+
    # |  a|       1|
    # |  b|       2|
    # |  c|       2|
    # |  d|       1|
    # |  e|       1|
    # |  f|       1|
    # +---+--------+
    
    g.outDegrees.sort("id").show()
    # +---+---------+
    # | id|outDegree|
    # +---+---------+
    # |  a|        2|
    # |  b|        1|
    # |  c|        1|
    # |  d|        1|
    # |  e|        2|
    # |  f|        1|
    # +---+---------+
    
    g.degrees.sort("id").show()
    # +---+------+
    # | id|degree|
    # +---+------+
    # |  a|     3|
    # |  b|     3|
    # |  c|     3|
    # |  d|     2|
    # |  e|     3|
    # |  f|     2|
    # +---+------+

## 对顶点和边的数据进行分析 ##

# 对边和顶点的数据进行分析
    agelist = g.vertices.groupBy().min("age").show()
    # +--------+
    # |min(age)|
    # +--------+
    # |      29|
    # +--------+
    
    youngest = g.vertices.filter("age < 30").show()
    # +---+-----+---+
    # | id| name|age|
    # +---+-----+---+
    # |  d|David| 29|
    # +---+-----+---+
    
    numFollows = g.edges.filter("relationship = 'follow'").count()
    print(numFollows)
    # 4

## 搜索指定结构路径 ##

## 对搜索结果过滤 ##

## 多路径条件搜索 ##

注意，搜索条件中，a和b可以指代相同的顶点，如果需要限制为不同的顶点，需要在返回结果中使用过滤器。

# 多个路径条件
    motif = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
    motif.show()
    # +----------------+--------------+----------------+--------------+
    # |               a|             e|               b|            e2|
    # +----------------+--------------+----------------+--------------+
    # |[c, Charlie, 30]|[c, b, follow]|    [b, Bob, 36]|[b, c, follow]|
    # |    [b, Bob, 36]|[b, c, follow]|[c, Charlie, 30]|[c, b, follow]|
    # +----------------+--------------+----------------+--------------+

## 匿名顶点和边 ##

# 不需要返回路径中的元素时，可以使用匿名顶点和边
    motif = g.find("(start)-[]->()")
    motif.show()
    # +----------------+
    # |           start|
    # +----------------+
    # |  [f, Fanny, 36]|
    # | [e, Esther, 32]|
    # | [e, Esther, 32]|
    # |  [d, David, 29]|
    # |[c, Charlie, 30]|
    # |    [b, Bob, 36]|
    # |  [a, Alice, 34]|
    # |  [a, Alice, 34]|
    # +----------------+

## 设置路径不存在的条件 ##

# 设置路径不存在的条件
    motif = g.find("(a)-[]->(b); !(b)-[]->(a)")
    motif.show()
    # +---------------+----------------+
    # |              a|               b|
    # +---------------+----------------+
    # | [a, Alice, 34]| [e, Esther, 32]|
    # |[e, Esther, 32]|  [d, David, 29]|
    # |[e, Esther, 32]|  [f, Fanny, 36]|
    # | [a, Alice, 34]|    [b, Bob, 36]|
    # | [f, Fanny, 36]|[c, Charlie, 30]|
    # | [d, David, 29]|  [a, Alice, 34]|
    # +---------------+----------------+

# 有状态和无状态查询 #

上面对路径的条件查询都是无状态的，没有指定任务限制条件，只能在查询完后再进行过滤。  
例如，我们要查询四个顶点的路径。

## 无状态查询 ##

使用无状态查询，例子如下：

from pyspark import SparkContext
    from pyspark.sql import SQLContext
    from graphframes.examples import Graphs
    
    sc = SparkContext("local", appName="mysqltest")
    sqlContext = SQLContext(sc)
    g = Graphs(sqlContext).friends()  # Get example graph
    g.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)").show()

返回一个较大的数据集：

## 有状态查询 ##

from pyspark import SparkContext
    from pyspark.sql import SQLContext
    from pyspark.sql.functions import col, lit, when
    
    from graphframes.examples import Graphs
    from functools import reduce
    
    
    sc = SparkContext("local", appName="mysqltest")
    sqlContext = SQLContext(sc)
    g = Graphs(sqlContext).friends()  # Get example graph
    chain4 = g.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)")
    
    
    # 定义下一个顶点更新状态的条件：如果关系为friend则cnt+1
    sumFriends = lambda cnt, relationship: when(relationship == "friend", cnt+1).otherwise(cnt)
    
    # 将更新方法应用到整个链的，链上每有一个关系是friend就加一，链上共三个关系。
    condition =reduce(lambda cnt, e: sumFriends(cnt, col(e).relationship), ["ab", "bc", "cd"], lit(0))
    
    # 传入限制条件
    chainWith2Friends2 = chain4.where(condition >= 2)
    chainWith2Friends2.show()

可以看到，我们在过程中限制的了链中必须有2个以上的关系是friend，结果符合预期。

+---------------+--------------+--------------+--------------+--------------+--------------+----------------+
    |              a|            ab|             b|            bc|             c|            cd|               d|
    +---------------+--------------+--------------+--------------+--------------+--------------+----------------+
    | [d, David, 29]|[d, a, friend]|[a, Alice, 34]|[a, b, friend]|  [b, Bob, 36]|[b, c, follow]|[c, Charlie, 30]|
    |[e, Esther, 32]|[e, d, friend]|[d, David, 29]|[d, a, friend]|[a, Alice, 34]|[a, b, friend]|    [b, Bob, 36]|
    +---------------+--------------+--------------+--------------+--------------+--------------+----------------+

# 子图 #

## 例一 ##

from graphframes.examples import Graphs
    from pyspark import SparkContext
    from pyspark.sql import SQLContext
    
    
    sc = SparkContext("local", appName="mysqltest")
    sqlContext = SQLContext(sc)
    g = Graphs(sqlContext).friends()  # Get example graph
    
    # 创建子图
    g1 = g.filterVertices("age > 30").filterEdges("relationship = 'friend'")
    g1.vertices.show()
    # +---+------+---+
    # | id|  name|age|
    # +---+------+---+
    # |  a| Alice| 34|
    # |  b|   Bob| 36|
    # |  e|Esther| 32|
    # |  f| Fanny| 36|
    # +---+------+---+
    
    
    g1 = g.filterVertices("age > 30").filterEdges("relationship = 'friend'").dropIsolatedVertices()
    g1.vertices.show()
    # +---+-----+---+
    # | id| name|age|
    # +---+-----+---+
    # |  b|  Bob| 36|
    # |  a|Alice| 34|
    # +---+-----+---+

这个例子中，我们使用filterVertices和filterEdges过滤顶点和边用来创建子图。  
在最后我们对子图调用dropIsolatedVertices()方法，删除孤立的没有连接的点。

## 例二 ##

from graphframes.examples import Graphs
    from pyspark import SparkContext
    from pyspark.sql import SQLContext
    from graphframes import GraphFrame
    
    
    sc = SparkContext("local", appName="mysqltest")
    sqlContext = SQLContext(sc)
    g = Graphs(sqlContext).friends()  # Get example graph
    
    # 路径搜索
    paths = g.find("(a)-[e]->(b)")\
      .filter("e.relationship = 'follow'")\
      .filter("a.age < b.age")
    paths.show()
    +----------------+--------------+--------------+
    # |               a|             e|             b|
    # +----------------+--------------+--------------+
    # | [e, Esther, 32]|[e, f, follow]|[f, Fanny, 36]|
    # |[c, Charlie, 30]|[c, b, follow]|  [b, Bob, 36]|
    # +----------------+--------------+--------------+
    
    # 选择关系数据集中的列
    e2 = paths.select("e.src", "e.dst", "e.relationship")
    # +---+---+------------+
    # |src|dst|relationship|
    # +---+---+------------+
    # |  e|  f|      follow|
    # |  c|  b|      follow|
    # +---+---+------------+
    
    # 使用顶点和边的集合构造子图
    g2 = GraphFrame(g.vertices, e2)

在上述例子的过程为：

1.  首先用`(a)-[e]->(b)`过滤出所有连接的顶点，得到顶点数据集。
2.  使用条件过滤指定关系follow和节点年龄的大小关系，得到一个边的数据集。这里可以看到，这个数据集中的元素，其实也是一个数据集。
3.  将顶点和边的数据集传入GraphFrame可以得到一个子图。

# 参考 #

> https://graphframes.github.io/graphframes/docs/\_site/user-guide.html  
> https://www.jianshu.com/p/d612fd2f4310  
> https://blog.csdn.net/ssyshenn/article/details/96702905