Как создать направленный граф с помощью Spark Graphx или Graphframe

Я пытаюсь запустить алгоритм связного компонента в моем наборе данных, но на направленном графике. Я не хочу, чтобы соединенный компонент был поперечным в обоих направлениях краев.

Это мой пример кода

import org.apache.log4j.{Level, LogManager}
import org.apache.spark.SparkConf
import org.apache.spark.graphx.Edge
import org.apache.spark.sql._
import org.graphframes._

object CCTest {

def main(args: Array[String]) {

  val sparkConf = new SparkConf()
      .setMaster("local[2]")
    .setAppName("cc_test")


  implicit val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

  val sc = sparkSession.sparkContext

  val vertex = sparkSession.createDataFrame(sc.parallelize(Array(
    (1L, "b4fcde907cbd290b7e16", 28),
    (2L, "cda389612d6b37674cb1", 27),
    (3L, "1a6a6e3fd2daaeeb2a05", 65),
    (4L, "9a007eee210a47e58047", 42),
    (5L, "e91898d39bf5f8501827", 55),
    (6L, "ceab58c59d23549d3f4b", 50),
    (12L, "ceab58c59asd3549d3f4b", 50),
    (14L, "ceab508c59d23549d3f4b", 55),
    (15L, "ceab508c59d23541d3f4b", 51)
  ))).toDF("id", "similar_hash", "size")

  val edges = sparkSession.createDataFrame(sc.parallelize(Array(
            Edge(2L, 1L, 0.7f),
            Edge(2L, 4L, 0.2f),
            Edge(3L, 2L, 0.4f),
            Edge(3L, 6L, 0.3f),
            Edge(4L, 1L, 0.1f),
            Edge(5L, 2L, 0.2f),
            Edge(5L, 3L, 0.8f),
            Edge(5L, 6L, 0.3f),
            Edge(12L, 14L, 1.3f),
            Edge(15L, 14L, 1.3f)    //< - should not be connected except (14L, 15L)
          ))).toDF("src", "dst", "attr")


  val graph = GraphFrame(vertex, edges)

  val cc = graph.connectedComponents.run()

  cc.show()



  sparkSession.stop()

}

 }

Результат:

+---+--------------------+----+---------+
| id|        similar_hash|size|component|
+---+--------------------+----+---------+
|  6|ceab58c59d23549d3f4b|  50|        1|
|  5|e91898d39bf5f8501827|  55|        1|
|  1|b4fcde907cbd290b7e16|  28|        1|
|  3|1a6a6e3fd2daaeeb2a05|  65|        1|
| 12|ceab58c59asd3549d...|  50|       12|
|  2|cda389612d6b37674cb1|  27|        1|
|  4|9a007eee210a47e58047|  42|        1|
| 14|ceab508c59d23549d...|  55|       12|
| 15|ceab508c59d23541d...|  51|       12|   <- should be in separate cluster
+---+--------------------+----+---------+

Как я могу этого добиться?


person Adetiloye Philip Kehinde    schedule 27.02.2017    source источник
comment
_ 1_.   -  person zero323    schedule 28.02.2017
comment
@ zero323 Я не думаю, что здесь можно использовать SCC (поправьте меня, если я ошибаюсь), поскольку он требует, чтобы каждый узел был доступен с других узлов. Но в основном мне нужно выполнить рекурсию, как рекурсивный запрос SQL. Например: a - ›b, b -› c, m - ›b Сильно CC: Set = {a, b}, Set = {b, c}, Set = {m, b} Подключенный компонент: это будет set = {a, b, d, m} - notice и m тоже подключен к nodea -> b -> m. Но в рекурсии a - ›b -› c = {a, b, c} и m - ›b = {m, b} это дает два набора {a, b, c} и {m, b}   -  person Adetiloye Philip Kehinde    schedule 28.02.2017
comment
OK. Так почему {a, b, c}, {m, b}, а не {a, b, c}, {m, b, c}? Так или иначе, похоже, для этого вам понадобится Pregel API.   -  person zero323    schedule 28.02.2017
comment
@ zero323 правильно - {a, b, c} и {m, b, c}. Я подумал об API Pregel. Я тоже попробую - Спасибо   -  person Adetiloye Philip Kehinde    schedule 28.02.2017
comment
Если можно, я постараюсь прояснить проблему. Похоже, вы больше ищете DFS или граф достижимости, чем подключенные компоненты.   -  person zero323    schedule 28.02.2017


Ответы (1)


Могу ошибиться, но мое решение напрямую связано с исходный код из connectedComponents. В строке 54 сама система вызывает только итератор Pregel, при этом ключевая строка

val pregelGraph = Pregel(ccGraph, initialMsg, maxIterations, EdgeDirection.Either)

Просто изменив EdgeDirection на более подходящий параметр (можно найти здесь), это может сработать для вас.

person dennlinger    schedule 15.03.2017