We used to rely on GraphX APIs but since we had so many efficiency problems and bugs, we stopped using GraphX. It's still being used in the GraphUtils though. If we choose to, remove all GraphX APIs and replace with internal case classes and whatnot. We should then have two sets of APIs, one internal to PageRank and one for general graph operations. I like this more since internal to PageRank, we need always tuples (for the joins), while outside of PageRank that makes the RDD's a bit messy with tuples all over the place. Could eventually consider Dataset/DataFrame APIs as well, but for now we stick to functional paradigms with RDDs.

When there are super nodes in the graph with hundreds of thousands or millions of out-edges, the PageRank join operation performs very poorly (stragglers) with skewed partitions.

Graph building: 0.2.0, 34 mins, 512 cores, 650M vertices, 14B edges
PageRank run: 0.2.0, 24 mins, 1024 cores, 650M vertices, 14B edges, 5 iterations
PageRank run: 0.2.0, 46 mins, 1024 cores, 650M vertices, 14B edges, 10 iterations

Only general edges/vertices stuff goes in here. Anything operating on the PageRankGraph as a whole should go in PageRankGraph instead.

This should probably be optional, somehow configured. Maybe you just want to do this on occassion (weekly?) or just during debugging, or maybe every n iterations?

Local checkpointing is not robust to executor failures, so we might lose information in the case executors go down and we don't have a redundant cache of lost block(s).

For both pull and publishing.

Depends on: #15

Improve documentation such that it is consumable as an open-source project. This should include any basic rationale about tuning, why or when to use this, examples, dataset sizes we have, etc.

Hello!
I have question about the result of the Vertice. the code is in the PageRank.scala file,
for example :

val actual = PageRank.run(
graph, teleportProb,
maxIterations,
convergenceThreshold
).collect()

Does the larger value indicates a stronger importance ?
if I use this value in black list,it indicates that
the person is more dangerous?Is it right?

They should sum to 1, maybe just add an external check and not necessarily inside the normalize edges function.

The options are: TSV, CSV, Parquet (or stay with object files). We need some interoperability with other formats in HDFS, like graph building reads from TSV in our case. What should be the standard interface to the apps? You can always build your own apps, the apps are just there for reference, handy usage/testing purposes.

/cc @maxjakob

This should provide some performance improvements over RDD's. Try it out and let's see.

This could just store object files, but should make it a simple one method call to do each. Then replace this stuff in the drivers.

See code and search for stuff.

Spark 2.2.1, patch versions of Scala, etc.

Re-run everything with the full datasets and validate that things are still performing after all of the refactoring.

Right now it's using an internal server.

sbt plugin to Sonatype: http://www.scala-sbt.org/0.13/docs/Using-Sonatype.html

Load the graph, but then replace the vertex values with those from a previous iteration on HDFS. This would require that the graph has the same vertices in each.

Use Travis or something like that.

This depends on all other tasks. Once complete and promoted, comment on the open Sonatype issue.

This should be couple of apps to both provide real use-case examples as well as for ongoing testing on real-sized graphs:

• remove hard coded paths
• show building a graph
• show running PageRank

This means removing all internal repos as well, or at least setting them up properly, only to be publish repos.

See: https://docs.travis-ci.com/user/languages/scala/

Depends on: #5

This seems to be great. Can you provide a few lines of code of a working example for this so it becomes easier to use for first-timers.

Thanks for the hep !

Right now we just support iteration number, and convergence as stopping criteria. This could be any function (in theory) including something like APA if the user is only concerned with stopping after the ranks are unchanged between iterations. Consider porting @maxjakob 's APA in here as a plugin option.

I think without specifying it, we are then just using POJO serialization. Not sure in Spark 2.1.0 how it works so time to check it out. It should also be easy for 3rd party users of our case classes to get this optimization, maybe just documented?

The cases are:

• new vertices in the new EdgesRDD that are not in the old VertexRDD
• vertices missing in the new EdgesRDD that are in the old VertexRDD

These need to be supported if you want to start from disk from a previous VertexRDD with old priors.

For every requirement of the graph structure, there should be a check/validation function as well as a function to correct any structure that needs it. Examples:

  /**
   * Removes any edges that are self-referencing the same vertex. That is, any
   * edges where the source and destination are the same. Any resulting vertices
   * that have no edges (in or out) will remain in the graph.
   */
  def removeSelfReferences(graph: Graph): Graph

  /**
   * Removes any vertices that have no in or out edges.
   */
  def removeDisconnectedVertices(graph: Graph): Graph

For open source adoption, it could be nice to have a drop-in replacement for the GraphX PageRank.

Use this since it's much more standard/openly available than our in-house version, and is better than the copy-paste we did to get started.

https://github.com/holdenk/spark-testing-base

Consider the package com.soundcloud.spark.graph so things don't collide with other internal Spark libraries.

This should be stand-alone but can be run on occasion to validate the assumptions and structure of the graph to run PageRank on. This should be a composed function of all the individual tests, and should be a guard function so it can be used in a pipeline.

  /**
   * Validates the structure of the input PageRank graph. See: {{#run}}.
   */
  def validateGraphStructure(edges: Edges, vertices: Vertices) {
    val numSelfReferences = countSelfReferences(graph.edges)
    val verticesAreNormalized = areVerticesNormalized(graph.vertices)
    val numVerticesWithoutNormalizedOutEdges = countVerticesWithoutNormalizedOutEdges(graph.edges)

    require(numSelfReferences == 0, "Number of vertices with self-referencing edges must be 0")
    require(verticesAreNormalized, "Input vertices values must be normalized")
    require(numVerticesWithoutNormalizedOutEdges == 0, "Number of vertices without normalized out edges must be 0")
  }

Read our internal guidelines to make sure we have everything covered

Along with the main README, we should ideally publish the Scaladocs in there as well. It's not immediately obvious how to use the README as the main index page and add docs so this is postponed to post-1.0.0 release.

See: https://alvinalexander.com/scala/how-to-generate-scala-documentation-scaladoc-command-examples
See: https://github.com/sbt/sbt-ghpages

Related: #9

Add local RDD checkpointing to truncate DAG/parents, after every iteration, this requires that dynamic allocation is off so also validate this at the start of the run by looking in the conf if possible (and document all this in the run function).

See: https://github.com/sbt/sbt-release#cross-building-during-a-release

• SoundCloud backstage blog post
• SoundCloud Dev Twitter
• Hacker News?

When performing a release, since we cross-compile, the first "push changes to repo" should be the only ones. The second time you get asked to tag and push to remote, you should choose to NOT do it since they versions are already updated. Furthermore, you should always make sure to specify the same release version, you will be asked twice (again, once per cross-compile version).