如何从 Spark 中的数据帧创建 EdgeRDD [英] how to create EdgeRDD from data frame in Spark

问题描述

我在 spark 中有一个数据框.每行代表一个人，我想检索他们之间可能的联系.拥有链接的规则是，对于每个可能的对，如果它们具有相同的 prop1:String 并且 prop2:Int 的绝对差小于5 则链接存在.我试图了解使用数据框完成此任务的最佳方法.

I have a dataframe in spark. Each row represents a person and I want to retrieve possible connections among them. The rule to have a link is that, for each possible pair, if they have the same prop1:String and the absolute difference of prop2:Int is < 5 then the link exists. I am trying to understand the best way to accomplish this task working with data frame.

我正在尝试检索索引的 RDD:

I am trying to retrieve indexed RDDs:

val idusers = people.select("ID")
                     .rdd
                     .map(r => r(0).asInstanceOf[Int])
                     .zipWithIndex
val prop1users = people.select("ID")
                        .rdd
                        .map(r => (r(0).asInstanceOf[Int], r(1).asInstanceOf[String]))
val prop2users = people.select("ID")
                        .rdd
                        .map(r => (r(0).asInstanceOf[Int], r(2).asInstanceOf[Int]))

然后开始删除重复项，例如:

then start removing duplicates like:

var links = idusers
            .join(idusers)
            .filter{ case (v1, v2) => v2._1 != v2._2 }

但后来我被困在检查 prop1 ......无论如何，有没有办法只使用数据框来完成所有这些步骤?

but then I got stuck to check for prop1... anyway, is there a way to accomplish all these steps just using data frame?

推荐答案

假设你有这样的事情:

val sqlc : SQLContext = ???

case class Person(id: Long, country: String, age: Int)

val testPeople = Seq(
  Person(1, "Romania"    , 15),
  Person(2, "New Zealand", 30),
  Person(3, "Romania"    , 17),
  Person(4, "Iceland"    , 20),
  Person(5, "Romania"    , 40),
  Person(6, "Romania"    , 44),
  Person(7, "Romania"    , 45),
  Person(8, "Iceland"    , 21),
  Person(9, "Iceland"    , 22)
)

val people = sqlc.createDataFrame(testPeople)

您可以通过重命名列来创建第一个自我奇迹，以避免在自联接中发生列冲突:

You can create first self miracle with columns renamed to avoid column-clashed in self-join:

val peopleR = people
  .withColumnRenamed("id"     , "idR")
  .withColumnRenamed("country", "countryR")
  .withColumnRenamed("age"    , "ageR")

现在您可以将数据帧与 self 连接起来，删除交换对和循环边:

Now you can join dataframe with self, dropping swapped pairs and loop-edges:

import org.apache.spark.sql.functions._

val relations = people.join(peopleR,
      (people("id") < peopleR("idR")) &&
        (people("country") === peopleR("countryR")) &&
        (abs(people("age") - peopleR("ageR")) < 5))

最后您可以构建所需的EdgeRDD:

Finally you can build desired EdgeRDD:

import org.apache.spark.graphx._

val edges = EdgeRDD.fromEdges(relations.map(row => Edge(
      row.getAs[Long]("id"), row.getAs[Long]("idR"), ())))

relations.show() 现在将输出:

+---+-------+---+---+--------+----+
| id|country|age|idR|countryR|ageR|
+---+-------+---+---+--------+----+
|  1|Romania| 15|  3| Romania|  17|
|  4|Iceland| 20|  8| Iceland|  21|
|  4|Iceland| 20|  9| Iceland|  22|
|  5|Romania| 40|  6| Romania|  44|
|  6|Romania| 44|  7| Romania|  45|
|  8|Iceland| 21|  9| Iceland|  22|
+---+-------+---+---+--------+----+

和 edges.toLocalIterator.foreach(println) 将输出:

Edge(1,3,())
Edge(4,8,())
Edge(4,9,())
Edge(5,6,())
Edge(6,7,())
Edge(8,9,())

这篇关于如何从 Spark 中的数据帧创建 EdgeRDD的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！