[jira] (SPARK-19098) Shuffled data leak/size doubling in ConnectedComponents/Pregel iterations

[Steven Ruppert (JIRA)]

Title: Message Title

Steven Ruppert commented on SPARK-19098   Re: Shuffled data leak/size doubling in ConnectedComponents/Pregel iterations The bug happens on spark 2.0.2 as well as 2.1.0, so it's not a regression. Instead, it appears to be a bad interaction between DataFrame-backed RDDs and GraphX. More specifically, the code I'm using is: val edges: RDD[GEdge[Int]] = spark.read.parquet("edges") .as[(VertexId, VertexId)] // Long .rdd .map { case (tail, head) => GEdge(tail, head, attr = 0) } val vertices: RDD[(VertexId, Int)] = spark.read.parquet("vertices") .as[VertexId] .map(vid => vid -> 0) // dummy vertex attr .rdd val graph = Graph(vertices, edges) val components: RDD[(VertexId, ComponentId)] = ConnectedComponents .run(graph, 10) However the VertexRDDImpl works, it drags something bad along when the vertex attributes are broadcaast. On a happier note, you can work around this bug by dropping the original DataFrame-backed rdds: val edges: RDD[GEdge[Int]] = spark.read.parquet("edges") .as[(VertexId, VertexId)] // Long .rdd .map { case (tail, head) => GEdge(tail, head, attr = 0) } val vertices: RDD[(VertexId, Int)] = spark.read.parquet("vertices") .as[VertexId] .map(vid => vid -> 0) // dummy vertex attr .rdd // XXX if you don't checkpoint, the parquet/dataframe-backed rdds // seem to "infect" the graphx rdds such that every pregel round // will double the amount of network traffic, adding up to terabytes. edges.checkpoint() vertices.checkpoint() val graph = Graph(vertices, edges) graph.checkpoint() // force checkpoints?? graph.degrees.count() val components: RDD[(VertexId, ComponentId)] = ConnectedComponents .run(graph, maxIterations = maxIterations.toInt) .vertices Once the graph is reading from checkpoints, everything works as expected, at least on spark 2.0.2. Add Comment   This message was sent by Atlassian JIRA (v6.3.15#6346-sha1:dbc023d)