apache spark graphx - 从 sql 表创建 VertexRDD

Question

我有一个要加载到 Spark 中 Dataframe 的表，它具有以下架构：

verticesDf.printSchema

root
 |-- id: integer (nullable = true)
 |-- target: string (nullable = true)
 |-- batch_id: integer (nullable = true)
 |-- x: double (nullable = true)
 |-- y: double (nullable = true)
 |-- z: double (nullable = true)
 |-- size: double (nullable = true)

如何将其转换为 VertexRDD，以便以后可以使用它构建 Graph？

我正在尝试以下方法：

case class SRow( target:String, batch_id:Double, x:Double, y:Double, z:Double, size:Double)
val sourceDS: Dataset[(VertexId, SRow)] = verticesDf.as[(VertexId, SRow)]
val vertVX=VertexRDD(sourceDS)

但是这个和许多其他的并没有给出结果——我总是遇到一些类型不匹配的问题。正确的方法是什么？

Answer 1

至少，要创建一个图表，您需要两个 RDD。包含顶点的RDD[(VertexId, VD)] 类型之一。 VertexId 只不过是 Long 和 VD 可以是任何东西，例如您的 Srow 类。另一个RDD是RDD[Edge[ED]]类型，其中ED类似于VD可以是任何东西。

这里讲一下vextex RDD的创建。您正在尝试将数据框转换为 Dataset[(VertexId, SRow)] 类型的数据集。它不起作用有两个原因。 id 是整数而不是长整数，结构错误。您的数据框包含两列以上。

这是怎么做的：

val vertices = verticesDf
    .select(
       // we transform the id to a long
       'id cast "long",
       // we create a struct with the other columns that will be turned into a Srow
       struct(verticesDf.columns.tail.map(col) : _*))
    .as[(Long, SRow)]

// we also need edges, let's create a dummy RDD
val edges = sc.parallelize(Seq(Edge(1L, 2L, "test")))

// And voila
val graph: Graph[SRow,String] = Graph(vertices.rdd, edges)

请注意最后一行，图是从 RDD 而非数据集创建的，因此我们需要对顶点进行转换。