-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #51 from tomerk/better-broadcasts
Added a broadcast via spark & switched to a single long-lived spark context
- Loading branch information
Showing
11 changed files
with
99 additions
and
103 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
4 changes: 2 additions & 2 deletions
4
veloxms-core/src/main/scala/edu/berkeley/veloxms/models/MatrixFactorizationModel.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
18 changes: 18 additions & 0 deletions
18
veloxms-core/src/main/scala/edu/berkeley/veloxms/storage/BroadcastProvider.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package edu.berkeley.veloxms.storage | ||
|
|
||
| import org.apache.spark.SparkContext | ||
|
|
||
| import scala.reflect.ClassTag | ||
|
|
||
| /** | ||
| * This trait provides Versioned Broadcasts. | ||
| * TODO: Add HTTP broadcast | ||
| * TODO: Add Torrent broadcast | ||
| */ | ||
| trait BroadcastProvider { | ||
| def get[T: ClassTag](id: String): VersionedBroadcast[T] | ||
| } | ||
|
|
||
| class SparkVersionedBroadcastProvider(sparkContext: SparkContext, path: String) extends BroadcastProvider { | ||
| override def get[T: ClassTag](id: String): VersionedBroadcast[T] = new SparkVersionedBroadcast(sparkContext, s"$path/$id") | ||
| } |
36 changes: 36 additions & 0 deletions
36
veloxms-core/src/main/scala/edu/berkeley/veloxms/storage/SparkVersionedBroadcast.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package edu.berkeley.veloxms.storage | ||
|
|
||
| import edu.berkeley.veloxms._ | ||
| import edu.berkeley.veloxms.util.{EtcdClient, KryoThreadLocal} | ||
| import org.apache.spark.{SparkContext, SparkConf} | ||
| import sun.misc.{BASE64Decoder, BASE64Encoder} | ||
|
|
||
| import scala.collection.mutable | ||
| import scala.reflect.ClassTag | ||
|
|
||
| /** | ||
| * A versioned broadcast that works via reading & writing to a global filesystem via the spark cluster | ||
| * @tparam T Has ClassTag because sc.objectFile (to load the broadcast) requires a classtag | ||
| */ | ||
| class SparkVersionedBroadcast[T: ClassTag](sc: SparkContext, path: String) extends VersionedBroadcast[T] { | ||
| private val cachedValues: mutable.Map[Version, T] = mutable.Map() | ||
|
|
||
| override def put(value: T, version: Version): Unit = this.synchronized { | ||
| sc.parallelize(Seq(value)).saveAsObjectFile(s"$path/$version") | ||
| } | ||
|
|
||
| override def get(version: Version): Option[T] = this.synchronized { | ||
| val out = cachedValues.get(version).orElse(fetch(version)) | ||
| out.foreach(x => cachedValues.put(version, x)) | ||
| out | ||
| } | ||
|
|
||
| override def cache(version: Version): Unit = this.synchronized { | ||
| fetch(version).foreach(x => cachedValues.put(version, x)) | ||
| } | ||
|
|
||
| private def fetch(version: Version): Option[T] = { | ||
| val location = s"$path/$version" | ||
| Some(sc.objectFile(location).first()) | ||
| } | ||
| } |
Oops, something went wrong.