假设你的数据是这样的
import org.apache.spark._
import org.graphframes._
val l = List(("Houston","Kyle","Benny"),("Houston","Benny","charles"),
("Houston","Charles","Denny"),("Omaha","carol","Brian"),
("Omaha","Brian","Daniel"),("Omaha","Sara","Marry"))
var df = spark.createDataFrame(l).toDF("city","src","dst")
创建要运行连接部件 cities = List("Houston","Omaha")
现在,在城市名单上运行的城市列的过滤器对每个城市的城市名单,然后从结果数据框中创建边和顶点数据帧。创建从这些边缘和顶点dataframes一个graphframe和运行连接组件的算法
val cities = List("Houston","Omaha")
for(city <- cities){
val edges = df.filter(df("city") === city).drop("city")
val vert = edges.select("src").union(edges.select("dst")).
distinct.select(col("src").alias("id"))
val g = GraphFrame(vert,edges)
val res = g.connectedComponents.run()
res.select("id", "component").orderBy("component").show()
}
输出
| id| component|
+-------+------------+
| Kyle|249108103168|
|charles|249108103168|
| Benny|249108103168|
|Charles|721554505728|
| Denny|721554505728|
+-------+------------+
+------+------------+
| id| component|
+------+------------+
| Marry|858993459200|
| Sara|858993459200|
| Brian|944892805120|
| carol|944892805120|
|Daniel|944892805120|
+------+------------+
避免问同样的问题两次:https://stackoverflow.com/questions/46386182/how-would -i-phrase-this-python-code-in-pyspark-sql-or-sql – Mariusz
删除旧的,这个有更好的措辞。 – oliver