Parent child pyspark 中的关系模型使用 Graphx/Spark

Question

我有一个 data-set，其中包含 (child, parent) 个实体。我需要从 data-set 中找到每个 child 的最终 parent。我的 data-set 有 130 万条记录。示例数据如下。

c-1, p-1
p-1, p-2
p-2, p-3
p-3, p-4

上面的样本数据中c-1的最终parent是p-4，p-1的最终parent是p-4等等。有时要找到 child 的最终 parent 我需要递归遍历多个级别。这是我到目前为止尝试过的。

我尝试创建一个 spark DF 并尝试递归地找到 parent 每个 child。但是这种方法需要很长时间。
我尝试创建可以应用于 data-set 的每一行的 UDF。但是我需要在 UDF 中调用 DF（查找 data-set）。但是火花没有支持在 UDF 中包含 DF。所以即使这种方法也没有帮助我。

关于如何解决这个问题有什么建议吗？

Answer 1

为了解决您提到的这两个问题，在 spark 中实施 CTE 是使用 Graphx Pregel API 可以解决您的问题。

下面是示例代码。

//setup & call the pregel api
def calcTopLevelHierarcy(vertexDF: DataFrame, edgeDF: DataFrame): RDD[(Any,(Int,Any,String,Int,Int))] = {

// create the vertex RDD
// primary key, root, path
val verticesRDD = vertexDF
  .rdd
  .map{x=> (x.get(0),x.get(1) , x.get(2))}
  .map{ x => (MurmurHash3.stringHash(x._1.toString).toLong, ( x._1.asInstanceOf[Any], x._2.asInstanceOf[Any] , x._3.asInstanceOf[String]) ) }

// create the edge RDD
// top down relationship
val EdgesRDD = edgeDF.rdd.map{x=> (x.get(0),x.get(1))}
  .map{ x => Edge(MurmurHash3.stringHash(x._1.toString).toLong,MurmurHash3.stringHash(x._2.toString).toLong,"topdown" )}

// create graph
val graph = Graph(verticesRDD, EdgesRDD).cache()

val pathSeperator = """/"""

// initialize id,level,root,path,iscyclic, isleaf
val initialMsg = (0L,0,0.asInstanceOf[Any],List("dummy"),0,1)

// add more dummy attributes to the vertices - id, level, root, path, isCyclic, existing value of current vertex to build path, isleaf, pk
val initialGraph = graph.mapVertices((id, v) => (id,0,v._2,List(v._3),0,v._3,1,v._1) )

val hrchyRDD = initialGraph.pregel(initialMsg,
  Int.MaxValue,
  EdgeDirection.Out)(
  setMsg,
  sendMsg,
  mergeMsg)


// build the path from the list
val hrchyOutRDD = hrchyRDD.vertices.map{case(id,v) => (v._8,(v._2,v._3,pathSeperator + v._4.reverse.mkString(pathSeperator),v._5, v._7 )) }

  hrchyOutRDD

}

在方法calcTopLevelHierarcy()中，你可以传入DataFrame（它解决了你的第二点）。

这是一个非常好的 link，带有一些示例代码。请看一下。

希望，这对您有所帮助。

Parent child pyspark 中的关系模型使用 Graphx/Spark

Parent child relationship model in pyspark using Graphx/Spark

pyspark

spark-graphx