8

I'm working with a largish (?) graph (60 million vertices and 9.5 billion edges) using Spark Graphframes. The underlying data is not large - the vertices take about 500mb on disk and the edges are about 40gb. My containers are frequently shutting down due to java heap out of memory problems, but I think the underlying problem is that the graphframe is constantly shuffling data around (I'm seeing shuffle read/write of up to 150gb). Is there a way to efficiently partition a Graphframe or the underlying edges/vertices to reduce shuffle?

John
  • 1,167
  • 1
  • 16
  • 33

2 Answers2

8

TL;DR It is not possible to efficiently partition Graphframe.

Graphframe algorithms can be separated into two categories:

  • Methods which delegate processing to GraphX counterpart. GraphX supports a number of partitioning methods but these are not exposed via Graphframe API. If you use one of these it is probably better to use GraphX directly.

    Unfortunately development of GraphX stopped almost completely with only a handful of small fixes over the last two years and overall performance is highly disappointing compared to both in-core libraries and out-of-core libraries.

  • Methods which are implemented natively using Spark Datasets, which considering limited programming model and only a single partitioning mode, are deeply unfit for complex graph processing.

    While relational columnar storage can be used for efficient graph processing naive iterative join approach employed by Graphframes just don't scale (but it is OK for shallow traversing with one or two hops).'

    You can try to repartition vertices and edges DataFrames by id and src respectively:

    val nPart: Int = ???

GraphFrame(v.repartition(nPart, v("id")), e.repartition(e(nPart, "src")))

    what should help in some cases.

Overall, at it's current (Dec, 2016) state, Spark is not a good choice for intensive graph analytics.

user7347764
  • 81
  • 2
  • 2
    Appreciate the insight. I found that I can boost performance by creating a custom column in the edge dataframe using the GraphX [partition schema](http://note.yuhc.me/2015/03/graphx-partition-strategy/) and partitioning on that. – John Jan 10 '17 at 21:46
  • 1
    @John I couldn't figure it out, do you have an example on how to do the partitioning on a custom column? – Philip K. Adetiloye Jan 26 '17 at 12:36
  • 4
    @John Can you share the update on How you resolved this performance issue ? – Choukha Ram Apr 30 '17 at 20:59
  • @John +1 for a coding example describing how you implemented the partition schema – Megan Nov 25 '19 at 19:41
1

Here's the partial solution / workaround - create a UDF that mimics one of the partition functions to create a new column and partition on that. 

num_parts = 256
random_vertex_cut = udf.register("random_vertex_cut", lambda src, dst: math.abs((src, dst).hashCode()) % num_parts, IntegerType())

edge.withColumn("v_cut", random_vertex_cut(col("src"), col("dst")).repartition(256, "v_cut")

This approach can help some, but not as well as GraphX.

John
  • 1,167
  • 1
  • 16
  • 33