Как построить график из фрейма данных? (GraphX)

#scala #apache-spark #dataframe #graph #spark-graphx

#scala #apache-spark #фрейм данных #График #spark-graphx

Вопрос:

Я новичок в scala и spark, и мне нужно построить график из фрейма данных. это структура моего фрейма данных, где S и O являются узлами, а столбец P представляет ребра.

  --------------------------- --------------------- ---------------------------- 
|S                          |P                    |O                           |
 --------------------------- --------------------- ---------------------------- 
|http://website/Jimmy_Carter|http://web/name      |James Earl Carter           |
|http://website/Jimmy_Car   |http://web/country   |http://website/United_States|
|http://website/Jimmy_Car   |http://web/birthPlace|http://web/Georgia_(US)     |
 --------------------------- --------------------- ----------------------------

Это код фрейма данных, и я хочу создать график из фрейма данных «dfA»

  val test = sc
     .textFile("testfile.ttl")
     .map(_.split(" "))
     .map(p => Triple(Try(p(0).toString()).toOption,
                      Try(p(1).toString()).toOption,
                      Try(p(2).toString()).toOption))
     .toDF()

  val url_regex = """^(?:"|<{1}s?)(.*)(?:>(?:s.)?|,s.*)$"""
  val dfA = test
      .withColumn("Subject", regexp_extract($"Subject", url_regex, 1))
      .withColumn("Predicate", regexp_extract($"Predicate", url_regex, 1))
      .withColumn("Object", regexp_extract($"Object", url_regex, 1))

1. У вас была возможность взглянуть на ответ? Решает ли это вашу проблему?

2. Привет, да, ответ решил проблему. но я изменил ‘val edges: RDD[Edge(VertexID, VertexID, строка)]’ на ‘val edges: RDD[Edge[Строка]] ‘. Большое спасибо!

3. И извините за поздний ответ, у меня возникли проблемы, и я не смог проверить код. Я сделал это сегодня.

Ответ №1:

Чтобы создать GraphX график, вам нужно извлечь вершины из вашего фрейма данных и связать их с идентификаторами. Затем вам нужно извлечь ребра (2 кортежа вершин метаданные), используя эти идентификаторы. И все это должно быть в RDDS, а не в фреймах данных.

Другими словами, вам нужно RDD[(VertexId, X)] для вершин и a, RDD[Edge(VertexId, VertexId, Y)] где X — метаданные вершины и Y метаданные ребра. Обратите внимание, что VertexId это просто псевдоним для Long .

В вашем случае, с «S» и «O» столбцами вершин и «P» столбцом ребер, это выглядело бы следующим образом.

 // Let's create the vertex RDD.
val vertices : RDD[(VertexId, String)] = df
    .select(explode(array('S, 'O))) // S and O are the vertices
    .distinct // we remove duplicates
    .rdd.map(_.getAs[String](0)) // transform to RDD
    .zipWithIndex // associate a long index to each vertex
    .map(_.swap)

// Now let's define a vertex dataframe because joins are clearer in sparkSQL
val vertexDf = vertices.toDF("id", "node")

// And let's extract the edges and join their vertices with their respective IDs
val edges : RDD[Edge(VertexId, VertexId, String)] = df
    .join(vertexDf, df("S") === vertexDf("node")) // getting the IDs for "S"
    .select('P, 'O, 'id as 'idS)
    .join(vertexDf, df("O") === vertexDf("node")) // getting the IDs for "O"
    .rdd.map(row => // creating the edge using column "P" as metadata 
      Edge(row.getAs[Long]("idS"), row.getAs[Long]("id"), row.getAs[String]("P")))

// And finally
val graph = Graph(vertices, edges)