Поиск пути между двумя вершинами, которые не связаны напрямую - PullRequest
Новая
       26

Поиск пути между двумя вершинами, которые не связаны напрямую

1 голос
/ 27 апреля 2019

У меня есть связанный граф, как этот 

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

Теперь я хочу найти всех пользователей, которые достигли от А до Е. Вывод должен быть user2, user3, user4 (так как все эти пользователи наконец достигли E из A, независимо от того, сколько прыжков они взяли). Как я могу написать мотив для этого. Это то, что я пытался. 

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 --packages graphframes:graphframes:0.3.0-spark2.0-s_2.11

Ответы [ 2 ]

1 голос
/ 28 апреля 2019

Мое предлагаемое решение довольно тривиально, но оно будет хорошо работать, если пути относительно короткие, а количество пользователей (т. Е. Количество строк в наборе данных) велико.Если это не так, пожалуйста, дайте мне знать, возможны другие реализации. 

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)
0 голосов
/ 13 мая 2019

Вы также можете использовать алгоритм 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]|
+------------+-------------+------------+-------------+---------+
...