如何从 Spark 中的数据框创建 EdgeRDD
how to create EdgeRDD from data frame in Spark
我在 spark 中有一个数据框。每行代表一个人,我想检索他们之间可能的联系。 link 的规则是,对于每个可能的对,如果它们具有相同的 prop1:String 并且 prop2:Int 的绝对差小于 5,则 link 存在。我正在尝试了解使用数据框完成此任务的最佳方法。
我正在尝试检索索引 RDD:
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]))
然后开始删除重复项,例如:
var links = idusers
.join(idusers)
.filter{ case (v1, v2) => v2._1 != v2._2 }
但后来我不得不检查 prop1...无论如何,有没有办法只使用数据框来完成所有这些步骤?
假设你有这样的事情:
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)
您可以通过重命名列来创建第一个自我奇迹,以避免在自连接中发生列冲突:
val peopleR = people
.withColumnRenamed("id" , "idR")
.withColumnRenamed("country", "countryR")
.withColumnRenamed("age" , "ageR")
现在你可以加入自己的数据帧,删除交换对和循环边:
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
:
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 中有一个数据框。每行代表一个人,我想检索他们之间可能的联系。 link 的规则是,对于每个可能的对,如果它们具有相同的 prop1:String 并且 prop2:Int 的绝对差小于 5,则 link 存在。我正在尝试了解使用数据框完成此任务的最佳方法。
我正在尝试检索索引 RDD:
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]))
然后开始删除重复项,例如:
var links = idusers
.join(idusers)
.filter{ case (v1, v2) => v2._1 != v2._2 }
但后来我不得不检查 prop1...无论如何,有没有办法只使用数据框来完成所有这些步骤?
假设你有这样的事情:
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)
您可以通过重命名列来创建第一个自我奇迹,以避免在自连接中发生列冲突:
val peopleR = people
.withColumnRenamed("id" , "idR")
.withColumnRenamed("country", "countryR")
.withColumnRenamed("age" , "ageR")
现在你可以加入自己的数据帧,删除交换对和循环边:
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
:
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,())