1

我有一个像这样的连接图

user1|A,C,B
user2|A,E,B,A
user3|C,B,A,B,E
user4|A,C,B,E,B

其中 user 是属性名称，并遵循该特定用户的路径。例如对于

user1 the  path is A->C->B
user2: A->E->B->A
user3: C->B->A->B->E
user4: A->C->B->E->B

现在，我想找到从 A 到达 E 的所有用户。输出应该是 user2、user3、user4（因为所有这些用户最终从 A 到达 E，不管他们走了多少跳）。我该如何写这个主题。这是我尝试过的。

val vertices=spark.createDataFrame(List(("A","Billing"),("B","Devices"),("C","Payment"),("D","Data"),("E","Help"))).toDF("id","desc")

val edges = spark.createDataFrame(List(("A","C","user1"),
("C","B","user1"),
("A","E","user2"),
("E","B","user2"),
("B","A","user2"),
("C","B","user3"),
("B","A","user3"),
("A","B","user3"),
("B","E","user3"),
("A","C","user4"),
("C","B","user4"),
("B","E","user4"),
("E","B","user4"))).toDF("src","dst","user")

val pathAnalysis=GraphFrame(vertices,edges)
pathAnalysis.find("(a)-[]->();()-[]->();()-[]->(d)").filter("a.id='A'").filter("d.id='E'").distinct().show()

但是我遇到了这样的例外

org.apache.spark.sql.AnalysisException: Detected implicit cartesian product for INNER join between logical plans
Join Inner
:- Project [a#355]
:  +- Join Inner, (__tmp-4363599943432734077#353.src = a#355.id)
:     :- LocalRelation [__tmp-4363599943432734077#353]
:     +- Project [named_struct(id, _1#0, desc, _2#1) AS a#355]
:        +- Filter (named_struct(id, _1#0, desc, _2#1).id = A)
:           +- LocalRelation [_1#0, _2#1]
+- LocalRelation
and
LocalRelation [__tmp-1043886091038848698#371]
Join condition is missing or trivial.
Either: use the CROSS JOIN syntax to allow cartesian products between these
relations, or: enable implicit cartesian products by setting the configuration
variable spark.sql.crossJoin.enabled=true;

我不确定我的情况是否正确或如何 spark.sql.crossJoin.enabled=true在 spark-shell 上设置此属性

我按如下方式调用了我的 spark-shell

spark-shell --packages graphframes:graphframes:0.3.0-spark2.0-s_2.11
4

2 回答 2

1

我建议的解决方案有点微不足道，但如果路径相对较短并且用户数量（即数据集中的行数）很大，它会很好地工作。如果不是这种情况，请告诉我，其他实现是可能的。

case class UserPath(
  userId: String,
  path: List[String])
val dsUsers = Seq(
  UserPath("user1", List("A", "B", "C")), 
  UserPath("user2", List("A", "E", "B", "A")))
.doDF.as[UserPath]

def pathExists(up: UserPath): Option[String] = {
  val prefix = up.path.takeWhile(s => s != "A")
  val len = up.path.length
  if (up.path.takeRight(len - prefix.length).contains("E"))
    Some(up.userId)
  else
    None
}
// Users with path from A -> E.
dsUsers.map(pathExists).filter(opt => !opt.isEmpty)
于 2019-04-28T06:46:11.520 回答
0

您也可以使用 BFS 算法： http: //graphframes.github.io/graphframes/docs/_site/api/scala/index.html#org.graphframes.lib.BFS 使用您的数据模型，您将不得不迭代用户并为每个用户运行 BFS，如下所示：

scala> pathAnalysis.bfs.fromExpr($"id" === "A").toExpr($"id" === "E").edgeFilter($"user" === "user3").run().show
+------------+-------------+------------+-------------+---------+
|        from|           e0|          v1|           e1|       to|
+------------+-------------+------------+-------------+---------+
|[A, Billing]|[A, B, user3]|[B, Devices]|[B, E, user3]|[E, Help]|
+------------+-------------+------------+-------------+---------+
于 2019-05-13T15:13:37.240 回答