如何打印一个 val 到 PartitionBy

Question

我在 Apache Spark GraphX 中遇到一个问题，我主要尝试使用此方法对一个图进行分区：

graph.partitionBy(HDRF, 128)

HDRF 是一种分区方法，我想打印其中的一个 val，我尝试打印但它没有打印任何东西

/编辑/

package app

import org.apache.spark.graphx._
import org.apache.spark._
import org.apache.spark.rdd.RDD


/**
  * Main del sistema
  */

object Main{


  def main(args: Array[String]) {

    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("HDRF"))

    // mostra solo i log in caso di errore
    sc.setLogLevel("ERROR")

    //modifico il file di testo preso in ingresso
    val edges:RDD[Edge[String]]=
      sc.textFile("data/u1.base").map{ line =>
        val fields= line.split("\t")
        Edge(fields(0).toLong,fields(1).toLong,fields(2))
      }

    val graph: Graph[Any,String] =Graph.fromEdges(edges,"defaultProperty")

    graph.partitionBy(HDRF,128)


  }
}

.

package app

import org.apache.spark.graphx._
import scala.collection.concurrent.TrieMap

object HDRF extends PartitionStrategy{
  private var init=0; //lo puoi usare per controllare una fase di inizializzazione che viene eseguita solo la prima volta

  private var partitionsLoad:Array[Long] = Array.empty[Long] //carico (numero di archi) di ogni partizione
  private val vertexIdListPartitions: TrieMap[Long, List[Long]] = TrieMap() //lista di partizioni associate a ogni vertice
  private val vertexIdEdges: TrieMap[Long, Long] = TrieMap() //grado di ogni vertice

  private var edges = 0

  private var sum :Long= 0

  override def getPartition(src:VertexId,dst:VertexId,numParts:Int): PartitionID ={
    var valoreMax:Long =Int.MaxValue
    var partScarica:Int = -1
    var c:Int = 0
    if(init==0){
      init=1
      partitionsLoad=Array.fill[Long](numParts)(0)
    }


    //AGGIORNA IL GRADO CONOSCIUTO DEI VERTICI src E dst NELLA VARIABILE vertexIdEdges
    vertexIdEdges(src)=vertexIdEdges(src)+1
    vertexIdEdges(dst)=vertexIdEdges(dst)+1
    sum=vertexIdEdges(src) + vertexIdEdges(dst)

    //PARTIZIONA IL GRAFO
    if((!vertexIdListPartitions.contains(src))&&(!vertexIdListPartitions.contains(dst))){
      //NESSUNO DEI DUE VERTICI E' STATO MAI INSERITO IN QUALCHE PARTIZIONE
      //SCELGO LA PARTZIIONE PIU' SCARICA E LI ASSEGNO A QUELLA
      while(c==numParts){
        if(partitionsLoad(c)<valoreMax){
          valoreMax=partitionsLoad(c)
          partScarica=c
        }
        c=c+1
      }
      if(partScarica != -1) {
        partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
        vertexIdListPartitions(partScarica).union(List(src, dst))
      }
      return partScarica

    }else if(((vertexIdListPartitions.contains(src))&&(!vertexIdListPartitions.contains(dst)))||((!vertexIdListPartitions.contains(src))&&(vertexIdListPartitions.contains(dst)))){
      //UNO SOLO DEI DUE VERTICI E' GIA' PRESENTE IN ALMENO UNA PARTIZIONE
      if((vertexIdListPartitions.contains(src))&&(!vertexIdListPartitions.contains(dst))){
        //SI TRATTA DI src
        //SCELGO LA PARTIZIONE PIU' SCARICA TRA QUELLE IN CUI E' PRESENTE src E CI REPLICO dst
        while(c==numParts){
          if(partitionsLoad(c)<valoreMax){
            if(vertexIdListPartitions(c).contains(src)) {
              valoreMax = partitionsLoad(c)
              partScarica = c
            }
          }
          c=c+1
        }
        if(partScarica != -1) {
          partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
          vertexIdListPartitions(partScarica).union(List(dst))
        }

      }else{
        //SI TRATTA DI dst
        //SCELGO LA PARTZIIONE PIU' SCARICA TRA QUELLE IN CUI E' PRESENTE dst E CI REPLICO src

        while(c==numParts){
          if(partitionsLoad(c)<valoreMax){
            if(vertexIdListPartitions(c).contains(src)) {
              valoreMax = partitionsLoad(c)
              partScarica = c
            }
          }
          c=c+1
        }
        if(partScarica != -1) {
          partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
          vertexIdListPartitions(partScarica).union(List(src))
        }

      }
    }else if(!vertexIdListPartitions(src).intersect(vertexIdListPartitions(dst)).isEmpty){
      //ENTRAMBI I VERTICI SONO PRESENTI IN DIVERSE PARTIZIONI ED ESISTE UNA INTERSEZIONE DEI SET NON NULLA (CIOE' ESISTE ALMENO UNA PARTIZIONE CHE LI CONTIENE ENTRAMBI)
      //SCELGO NELL'INTERSEZIONE DEI SET LA PARTIZIONE PIU' SCARICA

      while(c==numParts) {
        if (partitionsLoad(c) < valoreMax) {
          if (vertexIdListPartitions(c).contains(src) && vertexIdListPartitions(c).contains(dst)) {
            valoreMax = partitionsLoad(c)
            partScarica = c
          }
        }
        c = c + 1
      }
      if(partScarica != -1) {
        partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
        vertexIdListPartitions(partScarica).union(List(src))
      }

    }else {
      //ENTRAMBI I VERTICI SONO PRESENTI IN DIVERSE PARTIZIONI MA L'INTERSEZIONE DEI SET E' NULLA (CIOE' NON ESISTE ALCUNA PARTIZIONE CHE LI CONTIENE ENTRAMBI)
      if((vertexIdEdges(src))>=(vertexIdEdges(dst))){
        //SCELGO TRA LE PARTIZIONI A CUI E' ASSEGNATO dst QUELLA PIU' SCARICA E CI COPIO src

        while(c==numParts){
          if(partitionsLoad(c)<valoreMax){
            if(vertexIdListPartitions(c).contains(dst)) {
              valoreMax = partitionsLoad(c)
              partScarica = c
            }
          }
          c=c+1
        }
        if(partScarica != -1) {
          partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
          vertexIdListPartitions(partScarica).union(List(src))
        }

      }else{
        //SCELGO TRA LE PARTIZIONI A CUI E' ASSEGNATO src QUELLA PIU' SCARICA E CI COPIO dst

        while(c==numParts){
          if(partitionsLoad(c)<valoreMax){
            if(vertexIdListPartitions(c).contains(src)) {
              valoreMax = partitionsLoad(c)
              partScarica = c
            }
          }
          c=c+1
        }
        if(partScarica != -1) {
          partitionsLoad(partScarica) = partitionsLoad(partScarica) + 1
          vertexIdListPartitions(partScarica).union(List(dst))
        }
      }

    }
    edges=edges+1
    if(edges==80000) {
      print(sum)
    }
    return partScarica
  }
}

我需要打印总和，但我不明白为什么它没有出现。

Answer 1

partitionBy，与许多 Graph 函数一样，是一种惰性求值操作，它会生成一个新的 Graph 对象，但直到有必要时才会实际计算该图 - 即直到对结果执行某些 action（例如计数、持久化或收集）。

使用一个更简单的例子我们可以看到，如果我们对结果进行操作，这些打印将是可见的：

object SimpleExample extends PartitionStrategy {
  override def getPartition(src: VertexId, dst: VertexId, numParts: PartitionID): PartitionID = {
    println("partitioning!")
    numParts
  }
}

val result = graph.partitionBy(SimpleExample, 128) // nothing printed so far...

result.edges.count() // now that we act on the result, 
// we see "paritioning!" printed (several times).

注意从 PartitionStrategy 打印（或传递给 Spark 的任何转换函数将在 RDD、Graph 上执行, 或者 Dataset) 并没有太大帮助：这些函数在 工作节点 上执行，因此这些打印将 "scattered" 在不同进程的不同进程的输出中机器，并且可能在驱动程序应用程序（您的主要功能）的输出中不可见。

如何打印一个 val 到 PartitionBy

How to print one val to PartitionBy

scala

apache-spark

spark-graphx