GraphX pregel 和 spark 流：在 vprog 中推入 rddQueue 的 RDD 没有被处理答案

【问题标题】：GraphX pregel and spark streaming: the RDDs pushed into the rddQueue within the vprog are not processedGraphX pregel 和 spark 流：在 vprog 中推入 rddQueue 的 RDD 没有被处理
【发布时间】：2017-08-18 13:20:57
【问题描述】：

我正在使用 GraphX pregel 和 spark 流。我希望顶点程序（vprog）创建一个 RDD 并将其推送到 rddQueue 中进行处理。

val queueOfRDDs:Queue[RDD[Int]] = Queue.empty[RDD[Int]]        
@transient val streamingContext:StreamingContext = new StreamingContext(sc, Seconds(1))    
val inputDStream = streamingContext.queueStream(queueOfRDDs,true,null)
inputDStream.map(x => (x % 10, 1)).reduceByKey(_ + _).print()
streamingContext.start()

val initialMessage = "init"

def vertexProgram(id: VertexId, attr: String, msgs: String): String =
  {
    queueOfRDDs.synchronized {
      for(a <- 1 to 3) {
        queueOfRDDs.+=sc.makeRDD(1 to 1000, 10)
        println("will add " + queueOfRDDs.size)
      }
    }
    msgs
  }

  def sendMessage(...){...}
  def messageCombiner(...){...}
  val newGraph = Pregel.apply(graph,initialMessage,1,EdgeDirection.Out)(vertexProgram,sendMessage,messageCombiner)

预期结果是：

will add1
    will add2
    will add3
    will add4
    will add5
    will add6
    will add7
    ...
    -------------------------------------------
    Time: 1503048820000 ms
    -------------------------------------------
    (0,100)
    (6,100)
    (3,100)
    (9,100)
    (4,100)
    (1,100)
    (7,100)
    (8,100)
    (5,100)
    (2,100)

    -------------------------------------------
    Time: 1503048820000 ms
    -------------------------------------------
    (0,100)
    (6,100)
    (3,100)
    (9,100)
    (4,100)
    (1,100)
    (7,100)
    (8,100)
    (5,100)
    (2,100)

    ...

    -------------------------------------------
    Time: 1503048820000 ms
    -------------------------------------------
    (0,100)
    (6,100)
    (3,100)
    (9,100)
    (4,100)
    (1,100)
    (7,100)
    (8,100)
    (5,100)
    (2,100)

但我得到了这个结果：

    will add1
    will add2
    will add3
    will add4
    will add5
    will add6
    will add7
    ...

RDD 被推入 queueOfRDDs（它的大小增加了）但它们没有被处理。你能帮帮我吗

【问题讨论】：

标签： apache-spark spark-streaming spark-graphx

【解决方案1】：

TL;DR：这行不通。

此代码看起来不正确。看起来您正在尝试从任务 (vertexProgram) 中创建初始化 RDD，可能是通过使 SparkContext 变得懒惰或使用对象包装器。

您的程序附加到Queue 的本地副本，这对于实际的驱动程序是不可见的。即使是RDDs 也会对应不同的上下文。

【讨论】：