You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@spark.apache.org by Yifan LI <ia...@gmail.com> on 2014/09/03 17:58:09 UTC

Re: [GraphX] how to set memory configurations to avoid OutOfMemoryError "GC overhead limit exceeded"

Hi Ankur,

Thanks so much for your advice.

But it failed when I tried to set the storage level in constructing a graph.

val graph = GraphLoader.edgeListFile(sc, edgesFile, minEdgePartitions =
numPartitions).partitionBy(PartitionStrategy.EdgePartition2D).persist(StorageLevel.MEMORY_AND_DISK)

Error: java.lang.UnsupportedOperationException: Cannot change storage level
of an RDD after it was already assigned a level


Is there anyone could give me help?

Best,
Yifan




2014-08-18 23:52 GMT+02:00 Ankur Dave <an...@gmail.com>:

> On Mon, Aug 18, 2014 at 6:29 AM, Yifan LI <ia...@gmail.com> wrote:
>
>  I am testing our application(similar to "personalised page rank" using
>> Pregel, and note that each vertex property will need pretty much more space
>> to store after new iteration)
>
> [...]
>
> But when we ran it on larger graph(e.g. LiveJouranl), it always end at the
>> error "GC overhead limit exceeded", even the partitions number is increased
>> to 48 from 8.
>
>
> If the graph (including vertex properties) is too large to fit in memory,
> you might try allowing it to spill to disk. When constructing the graph,
> you can set vertexStorageLevel and edgeStorageLevel to
> StorageLevel.MEMORY_AND_DISK. This should allow the algorithm to finish.
>
> Ankur <http://www.ankurdave.com/>
>

Re: [GraphX] how to set memory configurations to avoid OutOfMemoryError "GC overhead limit exceeded"

Posted by Ankur Dave <an...@gmail.com>.
At 2014-09-05 12:13:18 +0200, Yifan LI <ia...@gmail.com> wrote:
> But how to assign the storage level to a new vertices RDD that mapped from
> an existing vertices RDD,
> e.g.
> *val newVertexRDD =
> graph.collectNeighborIds(EdgeDirection.Out).map{case(id:VertexId,
> a:Array[VertexId]) => (id, initialHashMap(a))}*
>
> the new one will be combined with that existing edges RDD(MEMORY_AND_DISK)
> to construct a new graph.
> e.g.
> val newGraph = Graph(newVertexRDD, graph.edges)

Sorry for the late reply. If you are constructing a graph from the derived VertexRDD, you can pass a desired storage level to the Graph constructor:

    val newVertexRDD = graph.collectNeighborIds(EdgeDirection.Out).map {
      case (id: VertexId, a: Array[VertexId]) => (id, initialHashMap(a))
    }
    val newGraph = Graph(
      newVertexRDD,
      graph.edges,
      edgeStorageLevel = StorageLevel.MEMORY_AND_DISK,
      vertexStorageLevel = StorageLevel.MEMORY_AND_DISK)

For others reading, the reason why GraphX needs to be told the desired storage level is that it internally constructs temporary vertex or edge RDDs and uses them more than once, so it has to cache them to avoid recomputation.

> BTW, the return of newVertexRDD.getStorageLevel is StorageLevel(true, true,
> false, true, 1), what does it mean?

See the StorageLevel object [1]. This particular storage level corresponds to StorageLevel.MEMORY_AND_DISK.

Ankur

[1] https://github.com/apache/spark/blob/092e2f152fb674e7200cc8a2cb99a8fe0a9b2b33/core/src/main/scala/org/apache/spark/storage/StorageLevel.scala#L147

---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org

Re: [GraphX] how to set memory configurations to avoid OutOfMemoryError "GC overhead limit exceeded"

Posted by Yifan LI <ia...@gmail.com>.
Thank you, Ankur! :)

But how to assign the storage level to a new vertices RDD that mapped from
an existing vertices RDD,
e.g.
*val newVertexRDD =
graph.collectNeighborIds(EdgeDirection.Out).map{case(id:VertexId,
a:Array[VertexId]) => (id, initialHashMap(a))}*

the new one will be combined with that existing edges RDD(MEMORY_AND_DISK)
to construct a new graph.
e.g.
val newGraph = Graph(newVertexRDD, graph.edges)


BTW, the return of newVertexRDD.getStorageLevel is StorageLevel(true, true,
false, true, 1), what does it mean?

Thanks in advance!

Best,
Yifan



2014-09-03 22:42 GMT+02:00 Ankur Dave <an...@gmail.com>:

> At 2014-09-03 17:58:09 +0200, Yifan LI <ia...@gmail.com> wrote:
> > val graph = GraphLoader.edgeListFile(sc, edgesFile, minEdgePartitions =
> numPartitions).partitionBy(PartitionStrategy.EdgePartition2D).persist(StorageLevel.MEMORY_AND_DISK)
> >
> > Error: java.lang.UnsupportedOperationException: Cannot change storage
> level
> > of an RDD after it was already assigned a level
>
> You have to pass the StorageLevel to GraphLoader.edgeListFile:
>
>     val graph = GraphLoader.edgeListFile(
>       sc, edgesFile, minEdgePartitions = numPartitions,
>       edgeStorageLevel = StorageLevel.MEMORY_AND_DISK,
>       vertexStorageLevel = StorageLevel.MEMORY_AND_DISK)
>       .partitionBy(PartitionStrategy.EdgePartition2D)
>
> Ankur
>

Re: [GraphX] how to set memory configurations to avoid OutOfMemoryError "GC overhead limit exceeded"

Posted by Ankur Dave <an...@gmail.com>.
At 2014-09-03 17:58:09 +0200, Yifan LI <ia...@gmail.com> wrote:
> val graph = GraphLoader.edgeListFile(sc, edgesFile, minEdgePartitions = numPartitions).partitionBy(PartitionStrategy.EdgePartition2D).persist(StorageLevel.MEMORY_AND_DISK)
>
> Error: java.lang.UnsupportedOperationException: Cannot change storage level
> of an RDD after it was already assigned a level

You have to pass the StorageLevel to GraphLoader.edgeListFile:

    val graph = GraphLoader.edgeListFile(
      sc, edgesFile, minEdgePartitions = numPartitions,
      edgeStorageLevel = StorageLevel.MEMORY_AND_DISK,
      vertexStorageLevel = StorageLevel.MEMORY_AND_DISK)
      .partitionBy(PartitionStrategy.EdgePartition2D)

Ankur

---------------------------------------------------------------------
To unsubscribe, e-mail: user-unsubscribe@spark.apache.org
For additional commands, e-mail: user-help@spark.apache.org