Spark Pregel이 Java와 함께 작동하지 않습니다.

Question

Java API를 사용하여 GraphX ​​및 Pregel에서 작업하고 있습니다. MaxValue 알고리즘을 구현하려고합니다 (주어진 가중 그래프 및 출력은 최대 가중치입니다). 하지만 내 구현이 작동하지 않습니다 :

public class Main { 

public static void main(String[] args){ 
    SparkConf conf = new SparkConf().setAppName("MaxValue").setMaster("spark://home:7077"); 

    JavaSparkContext sc = new JavaSparkContext(conf); 

    JavaRDD<String> text_file = sc.textFile(args[0]); 

    JavaRDD<String[]> text_file_arr = text_file.map(l -> l.split(" ")); 

    //cache 
    text_file_arr.cache(); 

    //create the vertex RDD 
    RDD<Tuple2<Object, Integer>> verteces = text_file_arr.map(
      t-> new Tuple2<>((Object) Long.parseLong(t[0]), Integer.parseInt(t[t.length-1])) 
    ).rdd(); 

    //create edge RDD 
    RDD<Edge<Boolean>> edges = text_file_arr 
      .flatMap(l -> { 
       List<Edge<Boolean>> edgeList = new ArrayList<>(); 
       long src = Long.parseLong(l[0]); 
       for (int i = 1;i<l.length-1;++i){ 
        edgeList.add(new Edge(src,Long.parseLong(l[i]),true)); 
       } 
       return edgeList.iterator(); 
      }) 
      .rdd(); 
    //create the graph 
    Graph<Integer,Boolean> graph = Graph.apply(
      verteces, 
      edges, 
      Integer.MIN_VALUE, 
      StorageLevel.MEMORY_AND_DISK(), 
      StorageLevel.MEMORY_AND_DISK(), 
      ClassTag$.MODULE$.apply(Integer.class), 
      ClassTag$.MODULE$.apply(Boolean.class) 
    ); 

    graph.edges().toJavaRDD().collect().forEach(System.out::print); 
    graph.vertices().toJavaRDD().collect().forEach(System.out::print); 

    GraphOps<Integer,Boolean> graph_ops = new GraphOps<>(
      graph, 
      ClassTag$.MODULE$.apply(Integer.class), 
      ClassTag$.MODULE$.apply(Boolean.class) 
    ); 
    //run pregel 
    Graph<Integer,Boolean> graph_pregel = graph_ops.pregel(
      Integer.MIN_VALUE, 
      3, 
      EdgeDirection.Either(), 
      new VProg(), 
      new SendMsg(), 
      new Merge(), 
      ClassTag$.MODULE$.apply(Integer.class) 
    ); 

    graph_pregel.vertices().toJavaRDD().saveAsTextFile("out"); 



    } 
} 

그리고 이것은 VProg, SendMsg 및 Merge 클래스입니다.

class SendMsg extends AbstractFunction1<EdgeTriplet<Integer,Boolean>, Iterator<Tuple2<Object, Integer>>> implements Serializable { 

    @Override 
    public Iterator<Tuple2<Object, Integer>> apply(EdgeTriplet<Integer, Boolean> et) { 
     System.out.println(et.srcId()+" ---> "+et.dstId()+" with: "+et.srcAttr()+" ---> "+et.dstId()); 

     if (et.srcAttr() > et.dstAttr()) { 
      return JavaConverters.asScalaIteratorConverter(Arrays.asList(et.toTuple()._1()).iterator()).asScala(); 
     }else{ 
      return JavaConverters.asScalaIteratorConverter(new ArrayList<Tuple2<Object, Integer>>().iterator()).asScala(); 
     } 
    } 
} 

class VProg extends AbstractFunction3<Object, Integer, Integer, Integer> implements Serializable{ 
    @Override 
    public Integer apply(Object l, Integer treeNodeThis, Integer treeNodeIn) { 
     if (treeNodeThis > treeNodeIn) { 
      System.out.println(l + " : " + treeNodeThis); 
      return treeNodeThis; 
     } else { 
      System.out.println(l + " : " + treeNodeIn); 
      return treeNodeIn; 
     } 
    } 
} 

class Merge extends AbstractFunction2<Integer, Integer, Integer> implements Serializable{ 
    @Override 
    public Integer apply(Integer n1, Integer n2) { 
     return (n1>n2)? n1:n2; 
    } 
} 

문제는 VPROG가 노드에서 실행 된 후 SendMsg가 하지만 값이 업데이트되지 않습니다 실행지고 있다는 것입니다. 이는 VProg이 새로운 값을 반환하지만 그래프는 여전히 입력 된 그래프라는 것을 의미합니다. 나 또한 다른 알고리즘을 시도하고 같은 문제가 생겼다. 어쩌면 내 수업 VProg, SendMsg 또는 Merge wrong?

그래프는 7 개의 노드로 연결되며 각 노드의 값은 2^nodenumber입니다.

또한, 내가 스파크 - 프리 젤에 버그가 생각 나는 많은 흔적 및 오류 후 스파크 2.0.0 및 Java 8

Answer 1

을 사용하고 ... 클래스의 프레 겔, 같은 문제로 시도 자바 API. 나는 스칼라와 같은 알고리즘을 구현하며 노력하고 있습니다 :

는

object Main { 
    def main(args: Array[String]): Unit = { 
    val conf = new SparkConf().setAppName("ScalaMaxValue").setMaster("spark://home:7077") 
    val sc = new SparkContext(conf) 

    val text_file_arr: RDD[Array[String]] = sc.textFile(args(0)).map(l => l.split(" ")) 

    val vertices: RDD[(VertexId, Int)] = text_file_arr.map(t => (t(0).toLong, t(t.length - 1).toInt)) 

    val edges: RDD[Edge[Boolean]] = text_file_arr.flatMap(l => { 
     val edgeList = new ListBuffer[Edge[Boolean]] //: List[Edge[Boolean]] = List() 
     val i = 0; 
     val src = l(0).toLong 
     for (i <- 0 to (l.length - 1)) { 
     val edge = Edge(src, l(i).toLong, true) 
     edgeList += edge 
     } 
     edgeList.toList 
    }); 

    val graph = Graph(vertices,edges,Int.MinValue) 

    val graph_pregel = Pregel(graph,Int.MinValue,Int.MaxValue)(vProg,sendMsg,merge) 

    //graph_pregel.vertices.saveAsTextFile("out") 

    println(graph_pregel.vertices.collect()(0)) 
    } 

    def vProg(id:VertexId, act: Int, other: Int): Int = { 
    if (other<act){ 
     act 
    }else{ 
     other 
    } 
    } 

    def sendMsg(et : EdgeTriplet[Int,Boolean]) : Iterator[(VertexId, Int)] = { 
    if(et.srcAttr > et.dstAttr){ 
     Iterator((et.dstId,et.srcAttr)) 
    }else{ 
     Iterator.empty 
    } 
    } 

    def merge(n1:Int, n2:Int): Int = { 
    if (n1<n2) n2 else n1 
    } 
} 

inputformat은 다음과 같습니다

#nodeID #neighborID_1 ... #neighborID_N #value 
. . .