Minor cleanup.
This commit is contained in:
parent
7134856351
commit
9764e579b8
|
@ -2,25 +2,9 @@ package spark.graph
|
||||||
|
|
||||||
import spark._
|
import spark._
|
||||||
import spark.SparkContext._
|
import spark.SparkContext._
|
||||||
// import com.esotericsoftware.kryo._
|
|
||||||
|
|
||||||
// import breeze.linalg._
|
// import breeze.linalg._
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// class AnalyticsKryoRegistrator extends KryoRegistrator {
|
|
||||||
// def registerClasses(kryo: Kryo) {
|
|
||||||
// println("registering kryo")
|
|
||||||
// kryo.register(classOf[(Int,Float,Float)])
|
|
||||||
// Graph.kryoRegister[(Int,Float,Float), Float](kryo)
|
|
||||||
// Graph.kryoRegister[(Int,Float), Float](kryo)
|
|
||||||
// Graph.kryoRegister[Int, Float](kryo)
|
|
||||||
// Graph.kryoRegister[Float, Float](kryo)
|
|
||||||
// kryo.setReferences(false);
|
|
||||||
// }
|
|
||||||
// }
|
|
||||||
|
|
||||||
|
|
||||||
object Analytics {
|
object Analytics {
|
||||||
|
|
||||||
|
|
||||||
|
@ -111,14 +95,6 @@ object Analytics {
|
||||||
gatherDirection = EdgeDirection.In)
|
gatherDirection = EdgeDirection.In)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
// /**
|
// /**
|
||||||
// * Compute the connected component membership of each vertex
|
// * Compute the connected component membership of each vertex
|
||||||
// * and return an RDD with the vertex value containing the
|
// * and return an RDD with the vertex value containing the
|
||||||
|
@ -250,9 +226,6 @@ object Analytics {
|
||||||
// //System.setProperty("spark.shuffle.compress", "false")
|
// //System.setProperty("spark.shuffle.compress", "false")
|
||||||
// System.setProperty("spark.kryo.registrator", "spark.graphlab.AnalyticsKryoRegistrator")
|
// System.setProperty("spark.kryo.registrator", "spark.graphlab.AnalyticsKryoRegistrator")
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
taskType match {
|
taskType match {
|
||||||
case "pagerank" => {
|
case "pagerank" => {
|
||||||
|
|
||||||
|
|
|
@ -43,12 +43,6 @@ class EdgeWithVertices[@specialized(Char, Int, Boolean, Byte, Long, Float, Doubl
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
private[graph]
|
|
||||||
case class MutableTuple2[@specialized(Char, Int, Boolean, Byte, Long, Float, Double) U,
|
|
||||||
@specialized(Char, Int, Boolean, Byte, Long, Float, Double) V](
|
|
||||||
var _1: U, var _2: V)
|
|
||||||
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A Graph RDD that supports computation on graphs.
|
* A Graph RDD that supports computation on graphs.
|
||||||
*/
|
*/
|
||||||
|
@ -61,26 +55,23 @@ class Graph[VD: ClassManifest, ED: ClassManifest] protected (
|
||||||
_rawETable: RDD[(Pid, EdgePartition[ED])]) {
|
_rawETable: RDD[(Pid, EdgePartition[ED])]) {
|
||||||
|
|
||||||
def this(vertices: RDD[Vertex[VD]], edges: RDD[Edge[ED]]) = {
|
def this(vertices: RDD[Vertex[VD]], edges: RDD[Edge[ED]]) = {
|
||||||
this(
|
this(vertices.partitions.size, edges.partitions.size, vertices, edges, null, null)
|
||||||
Graph.DEFAULT_NUM_VERTEX_PARTITIONS, Graph.DEFAULT_NUM_EDGE_PARTITIONS,
|
|
||||||
vertices, edges,
|
|
||||||
null, null)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
def withPartitioner(numVertexPartitions: Int, numEdgePartitions: Int): Graph[VD, ED] = {
|
def withPartitioner(numVertexPartitions: Int, numEdgePartitions: Int): Graph[VD, ED] = {
|
||||||
if (_cached) {
|
if (_cached) {
|
||||||
val newgraph = new Graph(numVertexPartitions, numEdgePartitions, null, null, _rawVTable, _rawETable)
|
(new Graph(numVertexPartitions, numEdgePartitions, null, null, _rawVTable, _rawETable))
|
||||||
newgraph.cache()
|
.cache()
|
||||||
} else {
|
} else {
|
||||||
new Graph(numVertexPartitions, numEdgePartitions, _rawVertices, _rawEdges, null, null)
|
new Graph(numVertexPartitions, numEdgePartitions, _rawVertices, _rawEdges, null, null)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
def withVertexPartitioner(numVertexPartitions: Int = Graph.DEFAULT_NUM_VERTEX_PARTITIONS) = {
|
def withVertexPartitioner(numVertexPartitions: Int) = {
|
||||||
withPartitioner(numVertexPartitions, numEdgePartitions)
|
withPartitioner(numVertexPartitions, numEdgePartitions)
|
||||||
}
|
}
|
||||||
|
|
||||||
def withEdgePartitioner(numEdgePartitions: Int = Graph.DEFAULT_NUM_EDGE_PARTITIONS) = {
|
def withEdgePartitioner(numEdgePartitions: Int) = {
|
||||||
withPartitioner(numVertexPartitions, numEdgePartitions)
|
withPartitioner(numVertexPartitions, numEdgePartitions)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -139,11 +130,11 @@ class Graph[VD: ClassManifest, ED: ClassManifest] protected (
|
||||||
edgeDirection)
|
edgeDirection)
|
||||||
}
|
}
|
||||||
|
|
||||||
def mapVertices[VD2: ClassManifest](f: (Vertex[VD]) => Vertex[VD2]): Graph[VD2, ED] = {
|
def mapVertices[VD2: ClassManifest](f: Vertex[VD] => Vertex[VD2]): Graph[VD2, ED] = {
|
||||||
newGraph(vertices.map(f), edges)
|
newGraph(vertices.map(f), edges)
|
||||||
}
|
}
|
||||||
|
|
||||||
def mapEdges[ED2: ClassManifest](f: (Edge[ED]) => Edge[ED2]): Graph[VD, ED2] = {
|
def mapEdges[ED2: ClassManifest](f: Edge[ED] => Edge[ED2]): Graph[VD, ED2] = {
|
||||||
newGraph(vertices, edges.map(f))
|
newGraph(vertices, edges.map(f))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -237,7 +228,7 @@ class Graph[VD: ClassManifest, ED: ClassManifest] protected (
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
vmap.int2ObjectEntrySet().fastIterator().filter{!_.getValue()._2.isEmpty}.map{ entry =>
|
vmap.int2ObjectEntrySet().fastIterator().filter(!_.getValue()._2.isEmpty).map{ entry =>
|
||||||
(entry.getIntKey(), entry.getValue()._2)
|
(entry.getIntKey(), entry.getValue()._2)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -316,14 +307,10 @@ class Graph[VD: ClassManifest, ED: ClassManifest] protected (
|
||||||
|
|
||||||
object Graph {
|
object Graph {
|
||||||
|
|
||||||
val DEFAULT_NUM_VERTEX_PARTITIONS = 5
|
|
||||||
val DEFAULT_NUM_EDGE_PARTITIONS = 5
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Load an edge list from file initializing the Graph RDD
|
* Load an edge list from file initializing the Graph RDD
|
||||||
*/
|
*/
|
||||||
def textFile[ED: ClassManifest](sc: SparkContext,
|
def textFile[ED: ClassManifest](sc: SparkContext, fname: String, edgeParser: Array[String] => ED) = {
|
||||||
fname: String, edgeParser: Array[String] => ED ) = {
|
|
||||||
|
|
||||||
// Parse the edge data table
|
// Parse the edge data table
|
||||||
val edges = sc.textFile(fname).map { line =>
|
val edges = sc.textFile(fname).map { line =>
|
||||||
|
|
|
@ -8,21 +8,17 @@ import spark.KryoRegistrator
|
||||||
class GraphKryoRegistrator extends KryoRegistrator {
|
class GraphKryoRegistrator extends KryoRegistrator {
|
||||||
|
|
||||||
def registerClasses(kryo: Kryo) {
|
def registerClasses(kryo: Kryo) {
|
||||||
kryo.register(classOf[(Int, Float, Float)])
|
//kryo.register(classOf[(Int, Float, Float)])
|
||||||
registerClass[(Int, Float, Float), Float](kryo)
|
registerClass[Int, Int, Int](kryo)
|
||||||
registerClass[(Int, Float), Float](kryo)
|
|
||||||
registerClass[Int, Float](kryo)
|
|
||||||
registerClass[Float, Float](kryo)
|
|
||||||
|
|
||||||
// This avoids a large number of hash table lookups.
|
// This avoids a large number of hash table lookups.
|
||||||
kryo.setReferences(false)
|
kryo.setReferences(false)
|
||||||
}
|
}
|
||||||
|
|
||||||
private def registerClass[VD: Manifest, ED: Manifest](kryo: Kryo) {
|
private def registerClass[VD: Manifest, ED: Manifest, VD2: Manifest](kryo: Kryo) {
|
||||||
//kryo.register(classManifest[VD].erasure)
|
kryo.register(classOf[Vertex[VD]])
|
||||||
// kryo.register(classManifest[ED].erasure)
|
kryo.register(classOf[Edge[ED]])
|
||||||
kryo.register(classOf[(Vid, Vid, ED)])
|
kryo.register(classOf[MutableTuple2[VD, VD2]])
|
||||||
kryo.register(classOf[(Vid, ED)])
|
kryo.register(classOf[(Vid, VD2)])
|
||||||
//kryo.register(classOf[EdgeBlockRecord[ED]])
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -11,4 +11,11 @@ package object graph {
|
||||||
* Return the default null-like value for a data type T.
|
* Return the default null-like value for a data type T.
|
||||||
*/
|
*/
|
||||||
def nullValue[T] = null.asInstanceOf[T]
|
def nullValue[T] = null.asInstanceOf[T]
|
||||||
|
|
||||||
|
|
||||||
|
private[graph]
|
||||||
|
case class MutableTuple2[@specialized(Char, Int, Boolean, Byte, Long, Float, Double) U,
|
||||||
|
@specialized(Char, Int, Boolean, Byte, Long, Float, Double) V](
|
||||||
|
var _1: U, var _2: V)
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue