You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2013/10/12 00:45:26 UTC
[01/50] git commit: Fixing closing brace indentation
Updated Branches:
refs/heads/scala-2.10 7d50f9f87 -> 0e5052ba3
Fixing closing brace indentation
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/6a7836cd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/6a7836cd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/6a7836cd
Branch: refs/heads/scala-2.10
Commit: 6a7836cddcf3ae0322e76cd5f79f6dd9ea73a09c
Parents: 0bd9b37
Author: Nick Pentreath <ni...@gmail.com>
Authored: Fri Oct 4 13:33:01 2013 +0200
Committer: Nick Pentreath <ni...@gmail.com>
Committed: Fri Oct 4 13:33:01 2013 +0200
----------------------------------------------------------------------
.../java/org/apache/spark/mllib/recommendation/JavaALSSuite.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/6a7836cd/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
----------------------------------------------------------------------
diff --git a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
index ec545ef..c8e59f7 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
@@ -90,8 +90,8 @@ public class JavaALSSuite implements Serializable {
double err = confidence * (truePref - prediction) * (truePref - prediction);
sqErr += err;
denom += 1.0;
- }
}
+ }
double rmse = Math.sqrt(sqErr / denom);
Assert.assertTrue(String.format("Confidence-weighted RMSE=%2.4f above threshold of %2.2f",
rmse, matchThreshold), Math.abs(rmse) < matchThreshold);
[41/50] git commit: Fixed comment to use javadoc style
Posted by ma...@apache.org.
Fixed comment to use javadoc style
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/3f7e9b26
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/3f7e9b26
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/3f7e9b26
Branch: refs/heads/scala-2.10
Commit: 3f7e9b265cfdd6ea2cb44e677b4e20186c3ae917
Parents: a34a4e8
Author: Kay Ousterhout <ka...@gmail.com>
Authored: Wed Oct 9 15:23:04 2013 -0700
Committer: Kay Ousterhout <ka...@gmail.com>
Committed: Wed Oct 9 15:23:04 2013 -0700
----------------------------------------------------------------------
.../org/apache/spark/scheduler/SparkListenerBus.scala | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/3f7e9b26/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
index 8283c4b..4d3e4a1 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
@@ -71,11 +71,11 @@ private[spark] class SparkListenerBus() extends Logging {
}
}
- /** Waits until there are no more events in the queue, or until the specified time has elapsed.
- *
- * Used for testing only. Returns true if the queue has emptied and false is the specified time
- * elapsed before the queue emptied.
- */
+ /**
+ * Waits until there are no more events in the queue, or until the specified time has elapsed.
+ * Used for testing only. Returns true if the queue has emptied and false is the specified time
+ * elapsed before the queue emptied.
+ */
def waitUntilEmpty(timeoutMillis: Int): Boolean = {
val finishTime = System.currentTimeMillis + timeoutMillis
while (!eventQueue.isEmpty()) {
[17/50] git commit: Bumping up test matrix size to eliminate random
failures
Posted by ma...@apache.org.
Bumping up test matrix size to eliminate random failures
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/b0f5f4d4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/b0f5f4d4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/b0f5f4d4
Branch: refs/heads/scala-2.10
Commit: b0f5f4d441119662c09572de697b2d9943f703ef
Parents: 93b96b4
Author: Nick Pentreath <ni...@gmail.com>
Authored: Mon Oct 7 11:44:22 2013 +0200
Committer: Nick Pentreath <ni...@gmail.com>
Committed: Mon Oct 7 11:44:22 2013 +0200
----------------------------------------------------------------------
.../spark/mllib/recommendation/JavaALSSuite.java | 16 ++++++++--------
.../spark/mllib/recommendation/ALSSuite.scala | 8 ++++----
2 files changed, 12 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/b0f5f4d4/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
----------------------------------------------------------------------
diff --git a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
index c8e59f7..eafee06 100644
--- a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
@@ -102,8 +102,8 @@ public class JavaALSSuite implements Serializable {
public void runALSUsingStaticMethods() {
int features = 1;
int iterations = 15;
- int users = 10;
- int products = 10;
+ int users = 50;
+ int products = 100;
scala.Tuple3<List<Rating>, DoubleMatrix, DoubleMatrix> testData = ALSSuite.generateRatingsAsJavaList(
users, products, features, 0.7, false);
@@ -116,8 +116,8 @@ public class JavaALSSuite implements Serializable {
public void runALSUsingConstructor() {
int features = 2;
int iterations = 15;
- int users = 20;
- int products = 30;
+ int users = 100;
+ int products = 200;
scala.Tuple3<List<Rating>, DoubleMatrix, DoubleMatrix> testData = ALSSuite.generateRatingsAsJavaList(
users, products, features, 0.7, false);
@@ -133,8 +133,8 @@ public class JavaALSSuite implements Serializable {
public void runImplicitALSUsingStaticMethods() {
int features = 1;
int iterations = 15;
- int users = 40;
- int products = 80;
+ int users = 80;
+ int products = 160;
scala.Tuple3<List<Rating>, DoubleMatrix, DoubleMatrix> testData = ALSSuite.generateRatingsAsJavaList(
users, products, features, 0.7, true);
@@ -147,8 +147,8 @@ public class JavaALSSuite implements Serializable {
public void runImplicitALSUsingConstructor() {
int features = 2;
int iterations = 15;
- int users = 50;
- int products = 100;
+ int users = 100;
+ int products = 200;
scala.Tuple3<List<Rating>, DoubleMatrix, DoubleMatrix> testData = ALSSuite.generateRatingsAsJavaList(
users, products, features, 0.7, true);
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/b0f5f4d4/mllib/src/test/scala/org/apache/spark/mllib/recommendation/ALSSuite.scala
----------------------------------------------------------------------
diff --git a/mllib/src/test/scala/org/apache/spark/mllib/recommendation/ALSSuite.scala b/mllib/src/test/scala/org/apache/spark/mllib/recommendation/ALSSuite.scala
index 1ab181d..fafc5ec 100644
--- a/mllib/src/test/scala/org/apache/spark/mllib/recommendation/ALSSuite.scala
+++ b/mllib/src/test/scala/org/apache/spark/mllib/recommendation/ALSSuite.scala
@@ -87,19 +87,19 @@ class ALSSuite extends FunSuite with BeforeAndAfterAll {
}
test("rank-1 matrices") {
- testALS(10, 20, 1, 15, 0.7, 0.3)
+ testALS(50, 100, 1, 15, 0.7, 0.3)
}
test("rank-2 matrices") {
- testALS(20, 30, 2, 15, 0.7, 0.3)
+ testALS(100, 200, 2, 15, 0.7, 0.3)
}
test("rank-1 matrices implicit") {
- testALS(40, 80, 1, 15, 0.7, 0.4, true)
+ testALS(80, 160, 1, 15, 0.7, 0.4, true)
}
test("rank-2 matrices implicit") {
- testALS(50, 100, 2, 15, 0.7, 0.4, true)
+ testALS(100, 200, 2, 15, 0.7, 0.4, true)
}
/**
[35/50] git commit: Use standard abbreviations in metrics labels
Posted by ma...@apache.org.
Use standard abbreviations in metrics labels
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/bd3bcc5f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/bd3bcc5f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/bd3bcc5f
Branch: refs/heads/scala-2.10
Commit: bd3bcc5f8e0db32679fe92705d0f7f3dd4360721
Parents: 19d445d
Author: Patrick Wendell <pw...@gmail.com>
Authored: Wed Oct 9 11:16:24 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Wed Oct 9 11:16:24 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/deploy/worker/WorkerSource.scala | 4 ++--
.../scala/org/apache/spark/storage/BlockManagerSource.scala | 8 ++++----
2 files changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/bd3bcc5f/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index b24e936..b7ddd8c 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -35,7 +35,7 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name("memUsed_MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memUsed_MB"), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
@@ -45,7 +45,7 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name("memFree_MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memFree_MB"), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/bd3bcc5f/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index e5068d5..365866d 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -28,7 +28,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", "maxMem_MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "maxMem_MB"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -36,7 +36,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "remainingMem_MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "remainingMem_MB"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -44,7 +44,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "memUsed_MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "memUsed_MB"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -53,7 +53,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed_MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed_MB"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
[12/50] git commit: Merge pull request #36 from pwendell/versions
Posted by ma...@apache.org.
Merge pull request #36 from pwendell/versions
Bumping EC2 default version in master to .
This change was already made on . This PR ports the change up to master.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/8fc68d04
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/8fc68d04
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/8fc68d04
Branch: refs/heads/scala-2.10
Commit: 8fc68d04bdea2a9bb895cb149b1b8e77d2ce0c19
Parents: 100222b 2484b84
Author: Reynold Xin <rx...@apache.org>
Authored: Sat Oct 5 17:24:35 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Sat Oct 5 17:24:35 2013 -0700
----------------------------------------------------------------------
ec2/spark_ec2.py | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
[07/50] git commit: Merge pull request #33 from
AndreSchumacher/pyspark_partition_key_change
Posted by ma...@apache.org.
Merge pull request #33 from AndreSchumacher/pyspark_partition_key_change
Fixing SPARK-602: PythonPartitioner
Currently PythonPartitioner determines partition ID by hashing a
byte-array representation of PySpark's key. This PR lets
PythonPartitioner use the actual partition ID, which is required e.g.
for sorting via PySpark.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/08641932
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/08641932
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/08641932
Branch: refs/heads/scala-2.10
Commit: 08641932bd17910cb5a839cdc7daeebfe4ae7ada
Parents: 232765f c84946f
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Sat Oct 5 13:25:18 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Sat Oct 5 13:25:18 2013 -0700
----------------------------------------------------------------------
.../apache/spark/api/python/PythonPartitioner.scala | 10 +++++++---
.../scala/org/apache/spark/api/python/PythonRDD.scala | 6 +++---
core/src/main/scala/org/apache/spark/util/Utils.scala | 13 +++++++++++++
.../test/scala/org/apache/spark/util/UtilsSuite.scala | 11 +++++++++++
python/pyspark/rdd.py | 10 ++++++----
python/pyspark/serializers.py | 4 ++++
6 files changed, 44 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
[33/50] git commit: Merge pull request #4 from MLnick/implicit-als
Posted by ma...@apache.org.
Merge pull request #4 from MLnick/implicit-als
Adding algorithm for implicit feedback data to ALS
This PR adds the commonly used "implicit feedack" variant to ALS.
The implementation is based in part on Mahout's implementation, which is in turn based on [Collaborative Filtering for Implicit Feedback Datasets](http://research.yahoo.com/pub/2433). It has been adapted for the blocked approach used in MLlib.
I have tested this implementation against the MovieLens 100k, 1m and 10m datasets, and confirmed that it produces the same RMSE score as Mahout, as well as my own port of Mahout's implicit ALS implementation to Spark (not that RMSE is necessarily the best metric to judge by for implicit feedback, but it provides a consistent metric for comparison).
It turned out to be more straightforward than I had thought to add this. The main additions are:
1. Adding `implicitPrefs` boolean flag and `alpha` parameter
2. Added the `computeYtY` method. In each least-squares step, the algorithm requires the computation of `YtY`, where `Y` is the {user, item} factor matrix. Since the factors are already block-distributed in an `RDD`, this is quite straightforward to compute but does add an extra operation over the explicit version (but only twice per iteration)
3. Finally the actual solve step in `updateBlock` boils down to:
* a multiplication of the `XtX` matrix by `alpha * rating`
* a multiplication of the `Xty` vector by `1 + alpha * rating`
* when solving for the factor vector, the implicit variant adds the `YtY` matrix to the LHS
4. Added `trainImplicit` methods in the `ALS` object
5. Added test cases for both Scala and Java - based on achieving a confidence-weighted RMSE score < 0.4 (this is taken from Mahout's test cases)
It would be great to get some feedback on this and have people test things out against some datasets (MovieLens and others and perhaps proprietary datasets) both locally and on a cluster if possible. I have not yet tested on a cluster but will try to do that soon.
I have tried to make things as efficient as possible but if there are potential improvements let me know.
The results of a run against ml-1m are below (note the vanilla RMSE scores will be very different from the explicit variant):
**INPUTS**
```
iterations=10
factors=10
lambda=0.01
alpha=1
implicitPrefs=true
```
**RESULTS**
```
Spark MLlib 0.8.0-SNAPSHOT
RMSE = 3.1544
Time: 24.834 sec
```
```
My own port of Mahout's ALS to Spark (updated to 0.8.0-SNAPSHOT)
RMSE = 3.1543
Time: 58.708 sec
```
```
Mahout 0.8
time ./factorize-movielens-1M.sh /path/to/ratings/ml-1m/ratings.dat
real 3m48.648s
user 6m39.254s
sys 0m14.505s
RMSE = 3.1539
```
Results of a run against ml-10m
```
Spark MLlib
RMSE = 3.1200
Time: 162.348 sec
```
```
Mahout 0.8
real 23m2.220s
user 43m39.185s
sys 0m25.316s
RMSE = 3.1187
```
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/3218fa79
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/3218fa79
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/3218fa79
Branch: refs/heads/scala-2.10
Commit: 3218fa795ff3ddee855772184aebe99098701d4f
Parents: e67d5b9 a5e58b8
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Tue Oct 8 23:44:55 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Tue Oct 8 23:44:55 2013 -0700
----------------------------------------------------------------------
docs/mllib-guide.md | 24 ++-
.../apache/spark/mllib/recommendation/ALS.scala | 199 ++++++++++++++++---
.../mllib/recommendation/JavaALSSuite.java | 85 ++++++--
.../spark/mllib/recommendation/ALSSuite.scala | 75 +++++--
4 files changed, 320 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
[26/50] git commit: Merge pull request #42 from
pwendell/shuffle-read-perf
Posted by ma...@apache.org.
Merge pull request #42 from pwendell/shuffle-read-perf
Fix inconsistent and incorrect log messages in shuffle read path
The user-facing messages generated by the CacheManager are currently wrong and somewhat misleading. This patch makes the messages more accurate. It also uses a consistent representation of the partition being fetched (`rdd_xx_yy`) so that it's easier for users to trace what is going on when reading logs.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/ea34c521
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/ea34c521
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/ea34c521
Branch: refs/heads/scala-2.10
Commit: ea34c521025d3408d44d45ab5c132fd9791794f6
Parents: 02f37ee 8b37771
Author: Reynold Xin <rx...@apache.org>
Authored: Mon Oct 7 20:45:58 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Mon Oct 7 20:45:58 2013 -0700
----------------------------------------------------------------------
.../main/scala/org/apache/spark/CacheManager.scala | 17 ++++++++---------
.../org/apache/spark/storage/BlockManager.scala | 12 +++++++++++-
2 files changed, 19 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
[13/50] git commit: Some comments regarding JobConf and InputFormat
caching for HadoopRDDs.
Posted by ma...@apache.org.
Some comments regarding JobConf and InputFormat caching for HadoopRDDs.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/6a2bbec5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/6a2bbec5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/6a2bbec5
Branch: refs/heads/scala-2.10
Commit: 6a2bbec5e3840cea5c128d521fe91050de8689db
Parents: 96929f2
Author: Harvey Feng <ha...@databricks.com>
Authored: Sat Oct 5 17:39:17 2013 -0700
Committer: Harvey Feng <ha...@databricks.com>
Committed: Sat Oct 5 17:53:58 2013 -0700
----------------------------------------------------------------------
.../apache/spark/deploy/SparkHadoopUtil.scala | 10 ++++++----
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 19 +++++++++++++++++++
2 files changed, 25 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/6a2bbec5/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
index f416b95..993ba6b 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
@@ -27,14 +27,16 @@ import org.apache.hadoop.mapred.JobConf
* Contains util methods to interact with Hadoop from spark.
*/
class SparkHadoopUtil {
- // A general map for metadata needed during HadoopRDD split computation (e.g., HadoopFileRDD uses
- // this to cache JobConfs).
+ // A general, soft-reference map for metadata needed during HadoopRDD split computation
+ // (e.g., HadoopFileRDD uses this to cache JobConfs and InputFormats).
private[spark] val hadoopJobMetadata = new MapMaker().softValues().makeMap[String, Any]()
- // Return an appropriate (subclass) of Configuration. Creating config can initializes some hadoop subsystems
+ // Return an appropriate (subclass) of Configuration. Creating config can initializes some hadoop
+ // subsystems
def newConfiguration(): Configuration = new Configuration()
- // add any user credentials to the job conf which are necessary for running on a secure Hadoop cluster
+ // Add any user credentials to the job conf which are necessary for running on a secure Hadoop
+ // cluster
def addCredentials(conf: JobConf) {}
def isYarnMode(): Boolean = { false }
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/6a2bbec5/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
index 51e5bb8..d3b3fff 100644
--- a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
@@ -52,8 +52,15 @@ class HadoopFileRDD[K, V](
override def getJobConf(): JobConf = {
if (HadoopRDD.containsCachedMetadata(jobConfCacheKey)) {
+ // getJobConf() has been called previously, so there is already a local cache of the JobConf
+ // needed by this RDD.
return HadoopRDD.getCachedMetadata(jobConfCacheKey).asInstanceOf[JobConf]
} else {
+ // Create a new JobConf, set the input file/directory paths to read from, and cache the
+ // JobConf (i.e., in a shared hash map in the slave's JVM process that's accessible through
+ // HadoopRDD.putCachedMetadata()), so that we only create one copy across multiple
+ // getJobConf() calls for this RDD in the local process.
+ // The caching helps minimize GC, since a JobConf can contain ~10KB of temporary objects.
val newJobConf = new JobConf(broadcastedConf.value.value)
FileInputFormat.setInputPaths(newJobConf, path)
HadoopRDD.putCachedMetadata(jobConfCacheKey, newJobConf)
@@ -112,10 +119,16 @@ class HadoopRDD[K, V](
protected def getJobConf(): JobConf = {
val conf: Configuration = broadcastedConf.value.value
if (conf.isInstanceOf[JobConf]) {
+ // A user-broadcasted JobConf was provided to the HadoopRDD, so always use it.
return conf.asInstanceOf[JobConf]
} else if (HadoopRDD.containsCachedMetadata(jobConfCacheKey)) {
+ // getJobConf() has been called previously, so there is already a local cache of the JobConf
+ // needed by this RDD.
return HadoopRDD.getCachedMetadata(jobConfCacheKey).asInstanceOf[JobConf]
} else {
+ // Create a JobConf that will be cached and used across this RDD's getJobConf() calls in the
+ // local process. The local cache is accessed through HadoopRDD.putCachedMetadata().
+ // The caching helps minimize GC, since a JobConf can contain ~10KB of temporary objects.
val newJobConf = new JobConf(broadcastedConf.value.value)
HadoopRDD.putCachedMetadata(jobConfCacheKey, newJobConf)
return newJobConf
@@ -126,6 +139,8 @@ class HadoopRDD[K, V](
if (HadoopRDD.containsCachedMetadata(inputFormatCacheKey)) {
return HadoopRDD.getCachedMetadata(inputFormatCacheKey).asInstanceOf[InputFormat[K, V]]
}
+ // Once an InputFormat for this RDD is created, cache it so that only one reflection call is
+ // done in each local process.
val newInputFormat = ReflectionUtils.newInstance(inputFormatClass.asInstanceOf[Class[_]], conf)
.asInstanceOf[InputFormat[K, V]]
if (newInputFormat.isInstanceOf[Configurable]) {
@@ -197,6 +212,10 @@ class HadoopRDD[K, V](
}
private[spark] object HadoopRDD {
+ /**
+ * The three methods below are helpers for accessing the local map, a property of the SparkEnv of
+ * the local process.
+ */
def getCachedMetadata(key: String) = SparkEnv.get.hadoop.hadoopJobMetadata.get(key)
def containsCachedMetadata(key: String) = SparkEnv.get.hadoop.hadoopJobMetadata.containsKey(key)
[04/50] git commit: Fixing SPARK-602: PythonPartitioner
Posted by ma...@apache.org.
Fixing SPARK-602: PythonPartitioner
Currently PythonPartitioner determines partition ID by hashing a
byte-array representation of PySpark's key. This PR lets
PythonPartitioner use the actual partition ID, which is required e.g.
for sorting via PySpark.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c84946fe
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c84946fe
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c84946fe
Branch: refs/heads/scala-2.10
Commit: c84946fe210069259f5d42ab8fd22a5ddae91d12
Parents: 232765f
Author: Andre Schumacher <sc...@icsi.berkeley.edu>
Authored: Fri Oct 4 11:56:47 2013 -0700
Committer: Andre Schumacher <sc...@icsi.berkeley.edu>
Committed: Fri Oct 4 11:56:47 2013 -0700
----------------------------------------------------------------------
.../apache/spark/api/python/PythonPartitioner.scala | 10 +++++++---
.../scala/org/apache/spark/api/python/PythonRDD.scala | 6 +++---
core/src/main/scala/org/apache/spark/util/Utils.scala | 13 +++++++++++++
.../test/scala/org/apache/spark/util/UtilsSuite.scala | 11 +++++++++++
python/pyspark/rdd.py | 10 ++++++----
python/pyspark/serializers.py | 4 ++++
6 files changed, 44 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/core/src/main/scala/org/apache/spark/api/python/PythonPartitioner.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonPartitioner.scala b/core/src/main/scala/org/apache/spark/api/python/PythonPartitioner.scala
index b090c6e..2be4e32 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonPartitioner.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonPartitioner.scala
@@ -17,12 +17,13 @@
package org.apache.spark.api.python
-import org.apache.spark.Partitioner
import java.util.Arrays
+
+import org.apache.spark.Partitioner
import org.apache.spark.util.Utils
/**
- * A [[org.apache.spark.Partitioner]] that performs handling of byte arrays, for use by the Python API.
+ * A [[org.apache.spark.Partitioner]] that performs handling of long-valued keys, for use by the Python API.
*
* Stores the unique id() of the Python-side partitioning function so that it is incorporated into
* equality comparisons. Correctness requires that the id is a unique identifier for the
@@ -30,6 +31,7 @@ import org.apache.spark.util.Utils
* function). This can be ensured by using the Python id() function and maintaining a reference
* to the Python partitioning function so that its id() is not reused.
*/
+
private[spark] class PythonPartitioner(
override val numPartitions: Int,
val pyPartitionFunctionId: Long)
@@ -37,7 +39,9 @@ private[spark] class PythonPartitioner(
override def getPartition(key: Any): Int = key match {
case null => 0
- case key: Array[Byte] => Utils.nonNegativeMod(Arrays.hashCode(key), numPartitions)
+ // we don't trust the Python partition function to return valid partition ID's so
+ // let's do a modulo numPartitions in any case
+ case key: Long => Utils.nonNegativeMod(key.toInt, numPartitions)
case _ => Utils.nonNegativeMod(key.hashCode(), numPartitions)
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala b/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
index ccd3833..1f8ad68 100644
--- a/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
+++ b/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
@@ -187,14 +187,14 @@ private class PythonException(msg: String) extends Exception(msg)
* This is used by PySpark's shuffle operations.
*/
private class PairwiseRDD(prev: RDD[Array[Byte]]) extends
- RDD[(Array[Byte], Array[Byte])](prev) {
+ RDD[(Long, Array[Byte])](prev) {
override def getPartitions = prev.partitions
override def compute(split: Partition, context: TaskContext) =
prev.iterator(split, context).grouped(2).map {
- case Seq(a, b) => (a, b)
+ case Seq(a, b) => (Utils.deserializeLongValue(a), b)
case x => throw new SparkException("PairwiseRDD: unexpected value: " + x)
}
- val asJavaPairRDD : JavaPairRDD[Array[Byte], Array[Byte]] = JavaPairRDD.fromRDD(this)
+ val asJavaPairRDD : JavaPairRDD[Long, Array[Byte]] = JavaPairRDD.fromRDD(this)
}
private[spark] object PythonRDD {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/core/src/main/scala/org/apache/spark/util/Utils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 886f071..f384875 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -70,6 +70,19 @@ private[spark] object Utils extends Logging {
return ois.readObject.asInstanceOf[T]
}
+ /** Deserialize a Long value (used for {@link org.apache.spark.api.python.PythonPartitioner}) */
+ def deserializeLongValue(bytes: Array[Byte]) : Long = {
+ // Note: we assume that we are given a Long value encoded in network (big-endian) byte order
+ var result = bytes(7) & 0xFFL
+ result = result + ((bytes(6) & 0xFFL) << 8)
+ result = result + ((bytes(5) & 0xFFL) << 16)
+ result = result + ((bytes(4) & 0xFFL) << 24)
+ result = result + ((bytes(3) & 0xFFL) << 32)
+ result = result + ((bytes(2) & 0xFFL) << 40)
+ result = result + ((bytes(1) & 0xFFL) << 48)
+ result + ((bytes(0) & 0xFFL) << 56)
+ }
+
/** Serialize via nested stream using specific serializer */
def serializeViaNestedStream(os: OutputStream, ser: SerializerInstance)(f: SerializationStream => Unit) = {
val osWrapper = ser.serializeStream(new OutputStream {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
index e2859ca..4684c8c 100644
--- a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
+++ b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala
@@ -20,6 +20,7 @@ package org.apache.spark.util
import com.google.common.base.Charsets
import com.google.common.io.Files
import java.io.{ByteArrayOutputStream, ByteArrayInputStream, FileOutputStream, File}
+import java.nio.{ByteBuffer, ByteOrder}
import org.scalatest.FunSuite
import org.apache.commons.io.FileUtils
import scala.util.Random
@@ -135,5 +136,15 @@ class UtilsSuite extends FunSuite {
FileUtils.deleteDirectory(tmpDir2)
}
+
+ test("deserialize long value") {
+ val testval : Long = 9730889947L
+ val bbuf = ByteBuffer.allocate(8)
+ assert(bbuf.hasArray)
+ bbuf.order(ByteOrder.BIG_ENDIAN)
+ bbuf.putLong(testval)
+ assert(bbuf.array.length === 8)
+ assert(Utils.deserializeLongValue(bbuf.array) === testval)
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/python/pyspark/rdd.py
----------------------------------------------------------------------
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index 58e1849..39c402b 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -29,7 +29,7 @@ from threading import Thread
from pyspark import cloudpickle
from pyspark.serializers import batched, Batch, dump_pickle, load_pickle, \
- read_from_pickle_file
+ read_from_pickle_file, pack_long
from pyspark.join import python_join, python_left_outer_join, \
python_right_outer_join, python_cogroup
from pyspark.statcounter import StatCounter
@@ -690,11 +690,13 @@ class RDD(object):
# form the hash buckets in Python, transferring O(numPartitions) objects
# to Java. Each object is a (splitNumber, [objects]) pair.
def add_shuffle_key(split, iterator):
+
buckets = defaultdict(list)
+
for (k, v) in iterator:
buckets[partitionFunc(k) % numPartitions].append((k, v))
for (split, items) in buckets.iteritems():
- yield str(split)
+ yield pack_long(split)
yield dump_pickle(Batch(items))
keyed = PipelinedRDD(self, add_shuffle_key)
keyed._bypass_serializer = True
@@ -831,8 +833,8 @@ class RDD(object):
>>> sorted(x.subtractByKey(y).collect())
[('b', 4), ('b', 5)]
"""
- filter_func = lambda tpl: len(tpl[1][0]) > 0 and len(tpl[1][1]) == 0
- map_func = lambda tpl: [(tpl[0], val) for val in tpl[1][0]]
+ filter_func = lambda (key, vals): len(vals[0]) > 0 and len(vals[1]) == 0
+ map_func = lambda (key, vals): [(key, val) for val in vals[0]]
return self.cogroup(other, numPartitions).filter(filter_func).flatMap(map_func)
def subtract(self, other, numPartitions=None):
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84946fe/python/pyspark/serializers.py
----------------------------------------------------------------------
diff --git a/python/pyspark/serializers.py b/python/pyspark/serializers.py
index fecacd1..54fed1c 100644
--- a/python/pyspark/serializers.py
+++ b/python/pyspark/serializers.py
@@ -67,6 +67,10 @@ def write_long(value, stream):
stream.write(struct.pack("!q", value))
+def pack_long(value):
+ return struct.pack("!q", value)
+
+
def read_int(stream):
length = stream.read(4)
if length == "":
[34/50] git commit: Merge pull request #22 from GraceH/metrics-naming
Posted by ma...@apache.org.
Merge pull request #22 from GraceH/metrics-naming
SPARK-900 Use coarser grained naming for metrics
see SPARK-900 Use coarser grained naming for metrics.
Now the new metric name is formatted as {XXX.YYY.ZZZ.COUNTER_UNIT}, XXX.YYY.ZZZ represents the group name, which can group several metrics under the same Ganglia view.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/19d445d3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/19d445d3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/19d445d3
Branch: refs/heads/scala-2.10
Commit: 19d445d37c38a35cc68da91f53a8780b89a1f8c9
Parents: 3218fa7 f7628e4
Author: Patrick Wendell <pw...@gmail.com>
Authored: Wed Oct 9 11:08:34 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Wed Oct 9 11:08:34 2013 -0700
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../apache/spark/deploy/worker/WorkerSource.scala | 10 +++++-----
.../apache/spark/executor/ExecutorSource.scala | 18 +++++++++---------
.../spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
.../apache/spark/storage/BlockManagerSource.scala | 8 ++++----
6 files changed, 27 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
[50/50] git commit: Merge pull request #51 from ScrapCodes/scala-2.10
Posted by ma...@apache.org.
Merge pull request #51 from ScrapCodes/scala-2.10
Scala 2.10
This is the update to latest master.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/0e5052ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/0e5052ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/0e5052ba
Branch: refs/heads/scala-2.10
Commit: 0e5052ba3783d8b1ca986d542a6fcb4fdb897b46
Parents: 7d50f9f bfbd7e5
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Fri Oct 11 15:45:16 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Fri Oct 11 15:45:16 2013 -0700
----------------------------------------------------------------------
README.md | 1 +
assembly/pom.xml | 20 +-
bagel/pom.xml | 6 +-
core/pom.xml | 13 +-
.../scala/org/apache/spark/CacheManager.scala | 21 +-
.../scala/org/apache/spark/SparkContext.scala | 47 ++-
.../scala/org/apache/spark/TaskEndReason.scala | 8 +-
.../org/apache/spark/api/java/JavaRDDLike.scala | 10 +
.../spark/api/python/PythonPartitioner.scala | 10 +-
.../org/apache/spark/api/python/PythonRDD.scala | 6 +-
.../org/apache/spark/deploy/JsonProtocol.scala | 3 +-
.../apache/spark/deploy/SparkHadoopUtil.scala | 12 +-
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../spark/deploy/master/MasterSource.scala | 6 +-
.../spark/deploy/worker/WorkerSource.scala | 10 +-
.../org/apache/spark/executor/Executor.scala | 28 +-
.../apache/spark/executor/ExecutorSource.scala | 18 +-
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 140 ++++++--
.../main/scala/org/apache/spark/rdd/RDD.scala | 38 +-
.../apache/spark/scheduler/DAGScheduler.scala | 8 +-
.../spark/scheduler/DAGSchedulerEvent.scala | 1 -
.../spark/scheduler/DAGSchedulerSource.scala | 10 +-
.../org/apache/spark/scheduler/JobLogger.scala | 1 -
.../scala/org/apache/spark/scheduler/Pool.scala | 121 +++++++
.../apache/spark/scheduler/Schedulable.scala | 46 +++
.../spark/scheduler/SchedulableBuilder.scala | 150 ++++++++
.../spark/scheduler/SchedulingAlgorithm.scala | 81 +++++
.../apache/spark/scheduler/SchedulingMode.scala | 29 ++
.../apache/spark/scheduler/SparkListener.scala | 1 -
.../spark/scheduler/SparkListenerBus.scala | 18 +
.../org/apache/spark/scheduler/StageInfo.scala | 2 +-
.../spark/scheduler/TaskDescription.scala | 37 ++
.../org/apache/spark/scheduler/TaskInfo.scala | 72 ++++
.../apache/spark/scheduler/TaskLocality.scala | 32 ++
.../org/apache/spark/scheduler/TaskResult.scala | 15 +-
.../apache/spark/scheduler/TaskScheduler.scala | 5 +-
.../spark/scheduler/TaskSchedulerListener.scala | 1 -
.../apache/spark/scheduler/TaskSetManager.scala | 48 +++
.../scheduler/cluster/ClusterScheduler.scala | 57 ++-
.../cluster/ClusterTaskSetManager.scala | 169 ++++-----
.../apache/spark/scheduler/cluster/Pool.scala | 121 -------
.../spark/scheduler/cluster/Schedulable.scala | 48 ---
.../scheduler/cluster/SchedulableBuilder.scala | 150 --------
.../scheduler/cluster/SchedulingAlgorithm.scala | 81 -----
.../scheduler/cluster/SchedulingMode.scala | 29 --
.../cluster/StandaloneClusterMessage.scala | 1 +
.../cluster/StandaloneSchedulerBackend.scala | 1 +
.../scheduler/cluster/TaskDescription.scala | 37 --
.../spark/scheduler/cluster/TaskInfo.scala | 72 ----
.../spark/scheduler/cluster/TaskLocality.scala | 32 --
.../scheduler/cluster/TaskResultGetter.scala | 124 +++++++
.../scheduler/cluster/TaskSetManager.scala | 51 ---
.../mesos/CoarseMesosSchedulerBackend.scala | 286 +++++++++++++++
.../cluster/mesos/MesosSchedulerBackend.scala | 345 +++++++++++++++++++
.../spark/scheduler/local/LocalScheduler.scala | 8 +-
.../scheduler/local/LocalTaskSetManager.scala | 21 +-
.../mesos/CoarseMesosSchedulerBackend.scala | 286 ---------------
.../scheduler/mesos/MesosSchedulerBackend.scala | 343 ------------------
.../spark/serializer/KryoSerializer.scala | 22 +-
.../org/apache/spark/storage/BlockManager.scala | 39 ++-
.../spark/storage/BlockManagerSource.scala | 8 +-
.../org/apache/spark/storage/MemoryStore.scala | 14 +-
.../apache/spark/ui/UIWorkloadGenerator.scala | 2 +-
.../org/apache/spark/ui/exec/ExecutorsUI.scala | 2 +-
.../org/apache/spark/ui/jobs/IndexPage.scala | 2 +-
.../spark/ui/jobs/JobProgressListener.scala | 4 +-
.../apache/spark/ui/jobs/JobProgressUI.scala | 5 +-
.../org/apache/spark/ui/jobs/PoolTable.scala | 3 +-
.../org/apache/spark/ui/jobs/StagePage.scala | 6 +-
.../org/apache/spark/ui/jobs/StageTable.scala | 3 +-
.../scala/org/apache/spark/util/AkkaUtils.scala | 4 +-
.../scala/org/apache/spark/util/Utils.scala | 13 +
.../org/apache/spark/DistributedSuite.scala | 12 -
.../scala/org/apache/spark/rdd/RDDSuite.scala | 39 ++-
.../spark/scheduler/DAGSchedulerSuite.scala | 4 +-
.../spark/scheduler/SparkListenerSuite.scala | 26 +-
.../cluster/ClusterSchedulerSuite.scala | 15 +-
.../cluster/ClusterTaskSetManagerSuite.scala | 58 +++-
.../cluster/TaskResultGetterSuite.scala | 113 ++++++
.../spark/serializer/KryoSerializerSuite.scala | 21 ++
.../scala/org/apache/spark/ui/UISuite.scala | 7 +
.../org/apache/spark/util/UtilsSuite.scala | 11 +
docs/_config.yml | 4 +-
docs/_layouts/global.html | 4 +-
docs/mllib-guide.md | 24 +-
docs/python-programming-guide.md | 2 +-
docs/running-on-yarn.md | 1 +
ec2/README | 2 +-
ec2/spark_ec2.py | 110 ++++--
examples/pom.xml | 28 +-
make-distribution.sh | 2 +-
mllib/pom.xml | 10 +-
.../apache/spark/mllib/recommendation/ALS.scala | 199 +++++++++--
.../mllib/recommendation/JavaALSSuite.java | 85 ++++-
.../spark/mllib/recommendation/ALSSuite.scala | 75 +++-
pom.xml | 6 +-
project/SparkBuild.scala | 22 +-
python/pyspark/rdd.py | 70 +++-
python/pyspark/serializers.py | 4 +
python/pyspark/shell.py | 2 +-
repl-bin/pom.xml | 12 +-
repl/pom.xml | 18 +-
.../org/apache/spark/repl/SparkILoopInit.scala | 2 +-
streaming/pom.xml | 7 +-
tools/pom.xml | 8 +-
yarn/pom.xml | 8 +-
.../org/apache/spark/deploy/yarn/Client.scala | 7 +-
.../spark/deploy/yarn/ClientArguments.scala | 6 +
.../spark/deploy/yarn/WorkerRunnable.scala | 5 +-
109 files changed, 2751 insertions(+), 1744 deletions(-)
----------------------------------------------------------------------
[46/50] git commit: Merge branch 'scala-2.10' of
github.com:ScrapCodes/spark into scala-2.10
Posted by ma...@apache.org.
Merge branch 'scala-2.10' of github.com:ScrapCodes/spark into scala-2.10
Conflicts:
core/src/main/scala/org/apache/spark/scheduler/cluster/ClusterTaskSetManager.scala
project/SparkBuild.scala
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/26860639
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/26860639
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/26860639
Branch: refs/heads/scala-2.10
Commit: 26860639c5fee7fc23db1e686f8eb202921e4314
Parents: 7d50f9f 7be7568
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Thu Oct 10 09:42:23 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Thu Oct 10 09:42:23 2013 +0530
----------------------------------------------------------------------
README.md | 1 +
assembly/pom.xml | 20 +-
bagel/pom.xml | 6 +-
core/pom.xml | 13 +-
.../scala/org/apache/spark/CacheManager.scala | 21 +-
.../scala/org/apache/spark/SparkContext.scala | 45 ++-
.../scala/org/apache/spark/TaskEndReason.scala | 8 +-
.../org/apache/spark/api/java/JavaRDDLike.scala | 10 +
.../spark/api/python/PythonPartitioner.scala | 10 +-
.../org/apache/spark/api/python/PythonRDD.scala | 6 +-
.../org/apache/spark/deploy/JsonProtocol.scala | 3 +-
.../apache/spark/deploy/SparkHadoopUtil.scala | 12 +-
.../org/apache/spark/executor/Executor.scala | 26 +-
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 140 ++++++--
.../main/scala/org/apache/spark/rdd/RDD.scala | 38 +-
.../apache/spark/scheduler/DAGScheduler.scala | 6 +-
.../spark/scheduler/DAGSchedulerEvent.scala | 1 -
.../org/apache/spark/scheduler/JobLogger.scala | 1 -
.../scala/org/apache/spark/scheduler/Pool.scala | 121 +++++++
.../apache/spark/scheduler/Schedulable.scala | 46 +++
.../spark/scheduler/SchedulableBuilder.scala | 150 ++++++++
.../spark/scheduler/SchedulingAlgorithm.scala | 81 +++++
.../apache/spark/scheduler/SchedulingMode.scala | 29 ++
.../apache/spark/scheduler/SparkListener.scala | 1 -
.../org/apache/spark/scheduler/StageInfo.scala | 2 +-
.../spark/scheduler/TaskDescription.scala | 37 ++
.../org/apache/spark/scheduler/TaskInfo.scala | 72 ++++
.../apache/spark/scheduler/TaskLocality.scala | 32 ++
.../org/apache/spark/scheduler/TaskResult.scala | 15 +-
.../apache/spark/scheduler/TaskScheduler.scala | 5 +-
.../spark/scheduler/TaskSchedulerListener.scala | 1 -
.../apache/spark/scheduler/TaskSetManager.scala | 48 +++
.../scheduler/cluster/ClusterScheduler.scala | 57 ++-
.../cluster/ClusterTaskSetManager.scala | 169 ++++-----
.../apache/spark/scheduler/cluster/Pool.scala | 121 -------
.../spark/scheduler/cluster/Schedulable.scala | 48 ---
.../scheduler/cluster/SchedulableBuilder.scala | 150 --------
.../scheduler/cluster/SchedulingAlgorithm.scala | 81 -----
.../scheduler/cluster/SchedulingMode.scala | 29 --
.../cluster/StandaloneClusterMessage.scala | 1 +
.../cluster/StandaloneSchedulerBackend.scala | 1 +
.../scheduler/cluster/TaskDescription.scala | 37 --
.../spark/scheduler/cluster/TaskInfo.scala | 72 ----
.../spark/scheduler/cluster/TaskLocality.scala | 32 --
.../scheduler/cluster/TaskResultGetter.scala | 124 +++++++
.../scheduler/cluster/TaskSetManager.scala | 51 ---
.../mesos/CoarseMesosSchedulerBackend.scala | 286 +++++++++++++++
.../cluster/mesos/MesosSchedulerBackend.scala | 345 +++++++++++++++++++
.../spark/scheduler/local/LocalScheduler.scala | 8 +-
.../scheduler/local/LocalTaskSetManager.scala | 21 +-
.../mesos/CoarseMesosSchedulerBackend.scala | 286 ---------------
.../scheduler/mesos/MesosSchedulerBackend.scala | 343 ------------------
.../org/apache/spark/storage/BlockManager.scala | 39 ++-
.../org/apache/spark/storage/MemoryStore.scala | 14 +-
.../apache/spark/ui/UIWorkloadGenerator.scala | 2 +-
.../org/apache/spark/ui/exec/ExecutorsUI.scala | 2 +-
.../org/apache/spark/ui/jobs/IndexPage.scala | 2 +-
.../spark/ui/jobs/JobProgressListener.scala | 4 +-
.../apache/spark/ui/jobs/JobProgressUI.scala | 5 +-
.../org/apache/spark/ui/jobs/PoolTable.scala | 3 +-
.../org/apache/spark/ui/jobs/StagePage.scala | 6 +-
.../org/apache/spark/ui/jobs/StageTable.scala | 3 +-
.../scala/org/apache/spark/util/Utils.scala | 13 +
.../org/apache/spark/DistributedSuite.scala | 1 +
.../scala/org/apache/spark/rdd/RDDSuite.scala | 39 ++-
.../spark/scheduler/DAGSchedulerSuite.scala | 4 +-
.../spark/scheduler/SparkListenerSuite.scala | 6 +-
.../cluster/ClusterSchedulerSuite.scala | 15 +-
.../cluster/ClusterTaskSetManagerSuite.scala | 58 +++-
.../cluster/TaskResultGetterSuite.scala | 113 ++++++
.../scala/org/apache/spark/ui/UISuite.scala | 7 +
.../org/apache/spark/util/UtilsSuite.scala | 11 +
docs/_config.yml | 4 +-
docs/_layouts/global.html | 4 +-
docs/running-on-yarn.md | 1 +
ec2/README | 2 +-
ec2/spark_ec2.py | 110 ++++--
examples/pom.xml | 28 +-
make-distribution.sh | 2 +-
mllib/pom.xml | 10 +-
pom.xml | 6 +-
project/SparkBuild.scala | 22 +-
python/pyspark/rdd.py | 10 +-
python/pyspark/serializers.py | 4 +
python/pyspark/shell.py | 2 +-
repl-bin/pom.xml | 12 +-
repl/pom.xml | 18 +-
.../org/apache/spark/repl/SparkILoopInit.scala | 2 +-
streaming/pom.xml | 7 +-
tools/pom.xml | 8 +-
yarn/pom.xml | 8 +-
.../org/apache/spark/deploy/yarn/Client.scala | 7 +-
.../spark/deploy/yarn/ClientArguments.scala | 6 +
.../spark/deploy/yarn/WorkerRunnable.scala | 5 +-
94 files changed, 2281 insertions(+), 1613 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/core/src/main/scala/org/apache/spark/scheduler/SchedulingMode.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/spark/scheduler/SchedulingMode.scala
index 0000000,0a786de..3832ee7
mode 000000,100644..100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SchedulingMode.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SchedulingMode.scala
@@@ -1,0 -1,29 +1,29 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ package org.apache.spark.scheduler
+
+ /**
+ * "FAIR" and "FIFO" determines which policy is used
+ * to order tasks amongst a Schedulable's sub-queues
+ * "NONE" is used when the a Schedulable has no sub-queues.
+ */
-object SchedulingMode extends Enumeration("FAIR", "FIFO", "NONE") {
++object SchedulingMode extends Enumeration {
+
+ type SchedulingMode = Value
+ val FAIR,FIFO,NONE = Value
+ }
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala
----------------------------------------------------------------------
diff --cc core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala
index 0000000,47b0f38..d31a5d5
mode 000000,100644..100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskLocality.scala
@@@ -1,0 -1,32 +1,32 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ package org.apache.spark.scheduler
+
+
+ private[spark] object TaskLocality
- extends Enumeration("PROCESS_LOCAL", "NODE_LOCAL", "RACK_LOCAL", "ANY")
++ extends Enumeration
+ {
+ // process local is expected to be used ONLY within tasksetmanager for now.
+ val PROCESS_LOCAL, NODE_LOCAL, RACK_LOCAL, ANY = Value
+
+ type TaskLocality = Value
+
+ def isAllowed(constraint: TaskLocality, condition: TaskLocality): Boolean = {
+ condition <= constraint
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/core/src/test/scala/org/apache/spark/DistributedSuite.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/26860639/project/SparkBuild.scala
----------------------------------------------------------------------
diff --cc project/SparkBuild.scala
index 5d4250a,2c63c0f..a734558
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@@ -79,9 -79,10 +79,10 @@@ object SparkBuild extends Build
def sharedSettings = Defaults.defaultSettings ++ Seq(
organization := "org.apache.spark",
- version := "0.8.0-SNAPSHOT",
+ version := "0.9.0-incubating-SNAPSHOT",
- scalaVersion := "2.10.2",
+ scalaVersion := "2.10.3",
- scalacOptions := Seq("-unchecked", "-optimize", "-deprecation", "-target:" + SCALAC_JVM_VERSION),
+ scalacOptions := Seq("-Xmax-classfile-name", "120", "-unchecked", "-deprecation",
+ "-target:" + SCALAC_JVM_VERSION),
javacOptions := Seq("-target", JAVAC_JVM_VERSION, "-source", JAVAC_JVM_VERSION),
unmanagedJars in Compile <<= baseDirectory map { base => (base / "lib" ** "*.jar").classpath },
retrieveManaged := true,
[44/50] git commit: Merge pull request #50 from
kayousterhout/SPARK-908
Posted by ma...@apache.org.
Merge pull request #50 from kayousterhout/SPARK-908
Fix race condition in SparkListenerSuite (fixes SPARK-908).
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/215238cb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/215238cb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/215238cb
Branch: refs/heads/scala-2.10
Commit: 215238cb399d46c83fafa64b3c98e0ebec21adb9
Parents: 7827efc 36966f6
Author: Reynold Xin <rx...@apache.org>
Authored: Wed Oct 9 16:49:44 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Wed Oct 9 16:49:44 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/SparkContext.scala | 2 +-
.../apache/spark/scheduler/DAGScheduler.scala | 2 +-
.../spark/scheduler/SparkListenerBus.scala | 18 +++++++++++
.../spark/scheduler/SparkListenerSuite.scala | 32 +++++++++++---------
4 files changed, 37 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
[28/50] git commit: Revert "remedy the line-wrap while exceeding 100
chars"
Posted by ma...@apache.org.
Revert "remedy the line-wrap while exceeding 100 chars"
This reverts commit 892fb8ffa85016a63d7d00dd6f1abc58ccf034a2.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a2af6b54
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a2af6b54
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a2af6b54
Branch: refs/heads/scala-2.10
Commit: a2af6b543a0a70d94a451c9022deea181d04f8e8
Parents: 892fb8f
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 17:44:56 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 17:44:56 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 13 ++--
.../spark/deploy/master/MasterSource.scala | 19 +++---
.../spark/deploy/worker/WorkerSource.scala | 31 +++++----
.../apache/spark/executor/ExecutorSource.scala | 45 +++++--------
.../spark/scheduler/DAGSchedulerSource.scala | 31 +++++----
.../spark/storage/BlockManagerSource.scala | 66 +++++++++-----------
6 files changed, 91 insertions(+), 114 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index f0b1f77..c72322e 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -31,13 +31,12 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")),
- new Gauge[Long] { override def getValue: Long = application.duration })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
+ override def getValue: Long = application.duration
+ })
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")),
- new Gauge[Int] { override def getValue: Int = application.coresGranted })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
+ override def getValue: Int = application.coresGranted
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index 8a88fef..de39398 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -27,18 +27,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("workers","number")),
- new Gauge[Int] { override def getValue: Int = master.workers.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
+ override def getValue: Int = master.workers.size
+ })
// Gauge for application numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")),
- new Gauge[Int] { override def getValue: Int = master.apps.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
+ override def getValue: Int = master.apps.size
+ })
// Gauge for waiting application numbers in cluster
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")),
- new Gauge[Int] { override def getValue: Int = master.waitingApps.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
+ override def getValue: Int = master.waitingApps.size
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index 0596f14..fc4f4ae 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -26,28 +26,27 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")),
- new Gauge[Int] { override def getValue: Int = worker.executors.size })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.executors.size
+ })
// Gauge for cores used of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")),
- new Gauge[Int] { override def getValue: Int = worker.coresUsed })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.coresUsed
+ })
// Gauge for memory used of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")),
- new Gauge[Int] { override def getValue: Int = worker.memoryUsed })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
+ override def getValue: Int = worker.memoryUsed
+ })
// Gauge for cores free of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")),
- new Gauge[Int] { override def getValue: Int = worker.coresFree })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
+ override def getValue: Int = worker.coresFree
+ })
// Gauge for memory free of this worker
- metricRegistry.register(
- MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")),
- new Gauge[Int] { override def getValue: Int = worker.memoryFree })
+ metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
+ override def getValue: Int = worker.memoryFree
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index d063e4a..6cbd154 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -44,42 +44,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getActiveCount() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getActiveCount()
+ })
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")),
- new Gauge[Long] { override def getValue: Long = executor.threadPool.getCompletedTaskCount() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
+ override def getValue: Long = executor.threadPool.getCompletedTaskCount()
+ })
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getPoolSize() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getPoolSize()
+ })
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(
- MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")),
- new Gauge[Int] { override def getValue: Int = executor.threadPool.getMaximumPoolSize() })
+ metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
+ override def getValue: Int = executor.threadPool.getMaximumPoolSize()
+ })
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("read", "bytes"),
- _.getBytesRead(), 0L)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("write", "bytes"),
- _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("read", "ops"),
- _.getReadOps(), 0)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("largeRead", "ops"),
- _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme,
- NamingConventions.makeMetricName("write", "ops"),
- _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
}
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 02fb807..9e90a08 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -28,24 +28,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.failed.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.failed.size
+ })
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.running.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.running.size
+ })
- metricRegistry.register(
- MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.waiting.size })
+ metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.waiting.size
+ })
- metricRegistry.register(
- MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.nextJobId.get() })
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.nextJobId.get()
+ })
- metricRegistry.register(
- MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")),
- new Gauge[Int] { override def getValue: Int = dagScheduler.activeJobs.size })
+ metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
+ override def getValue: Int = dagScheduler.activeJobs.size
+ })
}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a2af6b54/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index fcf9da4..4312250 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -29,48 +29,40 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- maxMem / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ maxMem / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- remainingMem / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ remainingMem / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")),
- new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
- val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
- (maxMem - remainingMem) / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
+ val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
+ (maxMem - remainingMem) / 1024 / 1024
+ }
})
- metricRegistry.register(
- MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
- override def getValue: Long = {
- val storageStatusList = blockManager.master.getStorageStatus
- val diskSpaceUsed = storageStatusList
- .flatMap(_.blocks.values.map(_.diskSize))
- .reduceOption(_ + _)
- .getOrElse(0L)
-
- diskSpaceUsed / 1024 / 1024
- }
+ metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
+ override def getValue: Long = {
+ val storageStatusList = blockManager.master.getStorageStatus
+ val diskSpaceUsed = storageStatusList
+ .flatMap(_.blocks.values.map(_.diskSize))
+ .reduceOption(_ + _)
+ .getOrElse(0L)
+
+ diskSpaceUsed / 1024 / 1024
+ }
})
}
-
[15/50] git commit: Merging build changes in from 0.8
Posted by ma...@apache.org.
Merging build changes in from 0.8
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/aa9fb849
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/aa9fb849
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/aa9fb849
Branch: refs/heads/scala-2.10
Commit: aa9fb849940bb7220286ce327f4fdb1706c6390e
Parents: 4a25b11
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sat Oct 5 17:53:11 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sat Oct 5 22:07:00 2013 -0700
----------------------------------------------------------------------
assembly/pom.xml | 18 ++++++------
bagel/pom.xml | 8 +++---
core/pom.xml | 15 +++++-----
.../spark/scheduler/SparkListenerSuite.scala | 6 +++-
docs/_layouts/global.html | 4 +--
examples/pom.xml | 30 ++++++++++++++------
mllib/pom.xml | 8 +++---
pom.xml | 7 +++--
repl-bin/pom.xml | 10 +++----
repl/pom.xml | 20 ++++++-------
streaming/pom.xml | 9 +++---
tools/pom.xml | 8 +++---
yarn/pom.xml | 6 ++--
13 files changed, 84 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/assembly/pom.xml
----------------------------------------------------------------------
diff --git a/assembly/pom.xml b/assembly/pom.xml
index d623321..09df8c1 100644
--- a/assembly/pom.xml
+++ b/assembly/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-assembly</artifactId>
+ <artifactId>spark-assembly_2.9.3</artifactId>
<name>Spark Project Assembly</name>
<url>http://spark.incubator.apache.org/</url>
@@ -41,27 +41,27 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
+ <artifactId>spark-bagel_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
+ <artifactId>spark-mllib_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
+ <artifactId>spark-repl_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
+ <artifactId>spark-streaming_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -104,13 +104,13 @@
</goals>
<configuration>
<transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/services/org.apache.hadoop.fs.FileSystem</resource>
</transformer>
</transformers>
<transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
@@ -128,7 +128,7 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-yarn</artifactId>
+ <artifactId>spark-yarn_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/bagel/pom.xml
----------------------------------------------------------------------
diff --git a/bagel/pom.xml b/bagel/pom.xml
index c4ce006..0e552c8 100644
--- a/bagel/pom.xml
+++ b/bagel/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
+ <artifactId>spark-bagel_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project Bagel</name>
<url>http://spark.incubator.apache.org/</url>
@@ -34,7 +34,7 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -43,12 +43,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index 9c2d604..d694508 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project Core</name>
<url>http://spark.incubator.apache.org/</url>
@@ -39,7 +39,6 @@
<dependency>
<groupId>net.java.dev.jets3t</groupId>
<artifactId>jets3t</artifactId>
- <version>0.7.1</version>
</dependency>
<dependency>
<groupId>org.apache.avro</groupId>
@@ -162,12 +161,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
@@ -202,14 +201,14 @@
<configuration>
<exportAntProperties>true</exportAntProperties>
<tasks>
- <property name="spark.classpath" refid="maven.test.classpath"/>
- <property environment="env"/>
+ <property name="spark.classpath" refid="maven.test.classpath" />
+ <property environment="env" />
<fail message="Please set the SCALA_HOME (or SCALA_LIBRARY_PATH if scala is on the path) environment variables and retry.">
<condition>
<not>
<or>
- <isset property="env.SCALA_HOME"/>
- <isset property="env.SCALA_LIBRARY_PATH"/>
+ <isset property="env.SCALA_HOME" />
+ <isset property="env.SCALA_LIBRARY_PATH" />
</or>
</not>
</condition>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index aac7c20..41a161e 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -29,7 +29,9 @@ import org.apache.spark.SparkContext._
class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatchers {
- test("local metrics") {
+ // TODO: This test has a race condition since the DAGScheduler now reports results
+ // asynchronously. It needs to be updated for that patch.
+ ignore("local metrics") {
sc = new SparkContext("local[4]", "test")
val listener = new SaveStageInfo
sc.addSparkListener(listener)
@@ -43,6 +45,7 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
val d = sc.parallelize(1 to 1e4.toInt, 64).map{i => w(i)}
d.count
+ Thread.sleep(1000)
listener.stageInfos.size should be (1)
val d2 = d.map{i => w(i) -> i * 2}.setName("shuffle input 1")
@@ -54,6 +57,7 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
d4.collectAsMap
+ Thread.sleep(1000)
listener.stageInfos.size should be (4)
listener.stageInfos.foreach {stageInfo =>
//small test, so some tasks might take less than 1 millisecond, but average should be greater than 1 ms
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/docs/_layouts/global.html
----------------------------------------------------------------------
diff --git a/docs/_layouts/global.html b/docs/_layouts/global.html
index 238ad26..0c1d657 100755
--- a/docs/_layouts/global.html
+++ b/docs/_layouts/global.html
@@ -6,7 +6,7 @@
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
- <title>{{ page.title }} - Spark {{site.SPARK_VERSION}} Documentation</title>
+ <title>{{ page.title }} - Spark {{site.SPARK_VERSION_SHORT}} Documentation</title>
<meta name="description" content="">
<link rel="stylesheet" href="css/bootstrap.min.css">
@@ -109,7 +109,7 @@
</ul>
</li>
</ul>
- <!--<p class="navbar-text pull-right"><span class="version-text">v{{site.SPARK_VERSION}}</span></p>-->
+ <!--<p class="navbar-text pull-right"><span class="version-text">v{{site.SPARK_VERSION_SHORT}}</span></p>-->
</div>
</div>
</div>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/examples/pom.xml
----------------------------------------------------------------------
diff --git a/examples/pom.xml b/examples/pom.xml
index b9cc6f5..b8c020a 100644
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@ -26,33 +26,41 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-examples</artifactId>
+ <artifactId>spark-examples_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project Examples</name>
<url>http://spark.incubator.apache.org/</url>
+ <repositories>
+ <!-- A repository in the local filesystem for the Kafka JAR, which we modified for Scala 2.9 -->
+ <repository>
+ <id>lib</id>
+ <url>file://${project.basedir}/lib</url>
+ </repository>
+ </repositories>
+
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
+ <artifactId>spark-streaming_2.9.3</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
+ <artifactId>spark-mllib_2.9.3</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
+ <artifactId>spark-bagel_2.9.3</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
@@ -72,6 +80,12 @@
</exclusions>
</dependency>
<dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka</artifactId>
+ <version>0.7.2-spark</version> <!-- Comes from our in-project repository -->
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-server</artifactId>
</dependency>
@@ -82,12 +96,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
@@ -161,7 +175,7 @@
</goals>
<configuration>
<transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/mllib/pom.xml
----------------------------------------------------------------------
diff --git a/mllib/pom.xml b/mllib/pom.xml
index 4ef4f0a..f472082 100644
--- a/mllib/pom.xml
+++ b/mllib/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
+ <artifactId>spark-mllib_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project ML Library</name>
<url>http://spark.incubator.apache.org/</url>
@@ -34,7 +34,7 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -48,12 +48,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index ad5051d..48bf38d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -40,6 +40,7 @@
<connection>scm:git:git@github.com:apache/incubator-spark.git</connection>
<developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/incubator-spark.git</developerConnection>
<url>scm:git:git@github.com:apache/incubator-spark.git</url>
+ <tag>HEAD</tag>
</scm>
<developers>
<developer>
@@ -322,7 +323,7 @@
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<version>1.9.1</version>
<scope>test</scope>
</dependency>
@@ -334,7 +335,7 @@
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<version>1.10.0</version>
<scope>test</scope>
</dependency>
@@ -603,7 +604,7 @@
<junitxml>.</junitxml>
<filereports>${project.build.directory}/SparkTestSuite.txt</filereports>
<argLine>-Xms64m -Xmx3g</argLine>
- <stderr/>
+ <stderr />
</configuration>
<executions>
<execution>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/repl-bin/pom.xml
----------------------------------------------------------------------
diff --git a/repl-bin/pom.xml b/repl-bin/pom.xml
index 05aadc7..f6bf94b 100644
--- a/repl-bin/pom.xml
+++ b/repl-bin/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl-bin</artifactId>
+ <artifactId>spark-repl-bin_2.9.3</artifactId>
<packaging>pom</packaging>
<name>Spark Project REPL binary packaging</name>
<url>http://spark.incubator.apache.org/</url>
@@ -40,18 +40,18 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
+ <artifactId>spark-bagel_2.9.3</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
+ <artifactId>spark-repl_2.9.3</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
@@ -89,7 +89,7 @@
</goals>
<configuration>
<transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/repl/pom.xml
----------------------------------------------------------------------
diff --git a/repl/pom.xml b/repl/pom.xml
index 2826c07..49d8662 100644
--- a/repl/pom.xml
+++ b/repl/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
+ <artifactId>spark-repl_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project REPL</name>
<url>http://spark.incubator.apache.org/</url>
@@ -39,18 +39,18 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
+ <artifactId>spark-bagel_2.9.3</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
+ <artifactId>spark-mllib_2.9.3</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
@@ -76,12 +76,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
@@ -101,14 +101,14 @@
<configuration>
<exportAntProperties>true</exportAntProperties>
<tasks>
- <property name="spark.classpath" refid="maven.test.classpath"/>
- <property environment="env"/>
+ <property name="spark.classpath" refid="maven.test.classpath" />
+ <property environment="env" />
<fail message="Please set the SCALA_HOME (or SCALA_LIBRARY_PATH if scala is on the path) environment variables and retry.">
<condition>
<not>
<or>
- <isset property="env.SCALA_HOME"/>
- <isset property="env.SCALA_LIBRARY_PATH"/>
+ <isset property="env.SCALA_HOME" />
+ <isset property="env.SCALA_LIBRARY_PATH" />
</or>
</not>
</condition>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/streaming/pom.xml
----------------------------------------------------------------------
diff --git a/streaming/pom.xml b/streaming/pom.xml
index b260a72..3b25fb4 100644
--- a/streaming/pom.xml
+++ b/streaming/pom.xml
@@ -26,7 +26,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
+ <artifactId>spark-streaming_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project Streaming</name>
<url>http://spark.incubator.apache.org/</url>
@@ -42,7 +42,7 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -58,6 +58,7 @@
<groupId>org.apache.kafka</groupId>
<artifactId>kafka</artifactId>
<version>0.7.2-spark</version> <!-- Comes from our in-project repository -->
+ <scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flume</groupId>
@@ -91,12 +92,12 @@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_${scala.version}</artifactId>
+ <artifactId>scalacheck_2.9.3</artifactId>
<scope>test</scope>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/tools/pom.xml
----------------------------------------------------------------------
diff --git a/tools/pom.xml b/tools/pom.xml
index 29f0014..f1c489b 100644
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@ -25,7 +25,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-tools</artifactId>
+ <artifactId>spark-tools_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project Tools</name>
<url>http://spark.incubator.apache.org/</url>
@@ -33,17 +33,17 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
+ <artifactId>spark-streaming_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_${scala.version}</artifactId>
+ <artifactId>scalatest_2.9.3</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/aa9fb849/yarn/pom.xml
----------------------------------------------------------------------
diff --git a/yarn/pom.xml b/yarn/pom.xml
index 427fcdf..3bc619d 100644
--- a/yarn/pom.xml
+++ b/yarn/pom.xml
@@ -25,7 +25,7 @@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-yarn</artifactId>
+ <artifactId>spark-yarn_2.9.3</artifactId>
<packaging>jar</packaging>
<name>Spark Project YARN Support</name>
<url>http://spark.incubator.apache.org/</url>
@@ -33,7 +33,7 @@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
+ <artifactId>spark-core_2.9.3</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@ -97,7 +97,7 @@
</goals>
<configuration>
<transformers>
- <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
+ <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
[14/50] git commit: Merge pull request #20 from
harveyfeng/hadoop-config-cache
Posted by ma...@apache.org.
Merge pull request #20 from harveyfeng/hadoop-config-cache
Allow users to pass broadcasted Configurations and cache InputFormats across Hadoop file reads.
Note: originally from https://github.com/mesos/spark/pull/942
Currently motivated by Shark queries on Hive-partitioned tables, where there's a JobConf broadcast for every Hive-partition (i.e., every subdirectory read). The only thing different about those JobConfs is the input path - the Hadoop Configuration that the JobConfs are constructed from remain the same.
This PR only modifies the old Hadoop API RDDs, but similar additions to the new API might reduce computation latencies a little bit for high-frequency FileInputDStreams (which only uses the new API right now).
As a small bonus, added InputFormats caching, to avoid reflection calls for every RDD#compute().
Few other notes:
Added a general soft-reference hashmap in SparkHadoopUtil because I wanted to avoid adding another class to SparkEnv.
SparkContext default hadoopConfiguration isn't cached. There's no equals() method for Configuration, so there isn't a good way to determine when configuration properties have changed.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/4a25b116
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/4a25b116
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/4a25b116
Branch: refs/heads/scala-2.10
Commit: 4a25b116d4e451afdf10fc4f018c383ed2c7789a
Parents: 8fc68d0 6a2bbec
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Sat Oct 5 19:28:55 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Sat Oct 5 19:28:55 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/CacheManager.scala | 4 +-
.../scala/org/apache/spark/SparkContext.scala | 39 ++++--
.../apache/spark/deploy/SparkHadoopUtil.scala | 12 +-
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 140 ++++++++++++++++---
4 files changed, 161 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/4a25b116/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
[27/50] git commit: Merge branch 'master' into wip-merge-master
Posted by ma...@apache.org.
Merge branch 'master' into wip-merge-master
Conflicts:
bagel/pom.xml
core/pom.xml
core/src/test/scala/org/apache/spark/ui/UISuite.scala
examples/pom.xml
mllib/pom.xml
pom.xml
project/SparkBuild.scala
repl/pom.xml
streaming/pom.xml
tools/pom.xml
In scala 2.10, a shorter representation is used for naming artifacts
so changed to shorter scala version for artifacts and made it a property in pom.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/7be75682
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/7be75682
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/7be75682
Branch: refs/heads/scala-2.10
Commit: 7be75682b931dd52014f3cfdc6887e54583ad0af
Parents: 3e41495 ea34c52
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Tue Oct 8 11:29:40 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Tue Oct 8 11:29:40 2013 +0530
----------------------------------------------------------------------
assembly/pom.xml | 18 +--
bagel/pom.xml | 4 +-
core/pom.xml | 11 +-
.../scala/org/apache/spark/CacheManager.scala | 21 +--
.../scala/org/apache/spark/SparkContext.scala | 39 ++++--
.../spark/api/python/PythonPartitioner.scala | 10 +-
.../org/apache/spark/api/python/PythonRDD.scala | 6 +-
.../org/apache/spark/deploy/JsonProtocol.scala | 3 +-
.../apache/spark/deploy/SparkHadoopUtil.scala | 12 +-
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 140 ++++++++++++++++---
.../org/apache/spark/storage/BlockManager.scala | 12 +-
.../scala/org/apache/spark/util/Utils.scala | 13 ++
.../spark/scheduler/SparkListenerSuite.scala | 6 +-
.../scala/org/apache/spark/ui/UISuite.scala | 7 +
.../org/apache/spark/util/UtilsSuite.scala | 11 ++
docs/_layouts/global.html | 4 +-
ec2/spark_ec2.py | 4 +-
examples/pom.xml | 26 +++-
mllib/pom.xml | 8 +-
pom.xml | 3 +-
project/SparkBuild.scala | 14 +-
python/pyspark/rdd.py | 10 +-
python/pyspark/serializers.py | 4 +
repl-bin/pom.xml | 10 +-
repl/pom.xml | 16 +--
streaming/pom.xml | 5 +-
tools/pom.xml | 6 +-
yarn/pom.xml | 6 +-
28 files changed, 314 insertions(+), 115 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/assembly/pom.xml
----------------------------------------------------------------------
diff --cc assembly/pom.xml
index f9c8b97,09df8c1..28b0692
--- a/assembly/pom.xml
+++ b/assembly/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-assembly</artifactId>
- <artifactId>spark-assembly_2.9.3</artifactId>
++ <artifactId>spark-assembly_${scala-short.version}</artifactId>
<name>Spark Project Assembly</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -41,27 -41,27 +41,27 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
- <artifactId>spark-bagel_2.9.3</artifactId>
++ <artifactId>spark-bagel_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
- <artifactId>spark-mllib_2.9.3</artifactId>
++ <artifactId>spark-mllib_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
- <artifactId>spark-repl_2.9.3</artifactId>
++ <artifactId>spark-repl_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
- <artifactId>spark-streaming_2.9.3</artifactId>
++ <artifactId>spark-streaming_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@@ -128,7 -128,7 +128,7 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-yarn</artifactId>
- <artifactId>spark-yarn_2.9.3</artifactId>
++ <artifactId>spark-yarn_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/bagel/pom.xml
----------------------------------------------------------------------
diff --cc bagel/pom.xml
index 271ab6c,0e552c8..c8b9c4f
--- a/bagel/pom.xml
+++ b/bagel/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
- <artifactId>spark-bagel_2.9.3</artifactId>
++ <artifactId>spark-bagel_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project Bagel</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -34,7 -34,7 +34,7 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/pom.xml
----------------------------------------------------------------------
diff --cc core/pom.xml
index 2ec8fa6,d694508..595240b
--- a/core/pom.xml
+++ b/core/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project Core</name>
<url>http://spark.incubator.apache.org/</url>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/src/main/scala/org/apache/spark/util/Utils.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/core/src/test/scala/org/apache/spark/ui/UISuite.scala
----------------------------------------------------------------------
diff --cc core/src/test/scala/org/apache/spark/ui/UISuite.scala
index f4e1d4e,8f0ec66..3764f4d
--- a/core/src/test/scala/org/apache/spark/ui/UISuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
@@@ -25,8 -25,16 +25,15 @@@ import org.eclipse.jetty.server.Serve
class UISuite extends FunSuite {
test("jetty port increases under contention") {
val startPort = 4040
+ val server = new Server(startPort)
+
+ Try { server.start() } match {
+ case Success(s) =>
+ case Failure(e) =>
+ // Either case server port is busy hence setup for test complete
+ }
val (jettyServer1, boundPort1) = JettyUtils.startJettyServer("localhost", startPort, Seq())
val (jettyServer2, boundPort2) = JettyUtils.startJettyServer("localhost", startPort, Seq())
-
// Allow some wiggle room in case ports on the machine are under contention
assert(boundPort1 > startPort && boundPort1 < startPort + 10)
assert(boundPort2 > boundPort1 && boundPort2 < boundPort1 + 10)
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/examples/pom.xml
----------------------------------------------------------------------
diff --cc examples/pom.xml
index 3c0a8d0,b8c020a..c6c9def
--- a/examples/pom.xml
+++ b/examples/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-examples</artifactId>
- <artifactId>spark-examples_2.9.3</artifactId>
++ <artifactId>spark-examples_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project Examples</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -34,25 -42,25 +42,25 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
- <artifactId>spark-streaming_2.9.3</artifactId>
++ <artifactId>spark-streaming_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
- <artifactId>spark-mllib_2.9.3</artifactId>
++ <artifactId>spark-mllib_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
- <artifactId>spark-bagel_2.9.3</artifactId>
++ <artifactId>spark-bagel_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/mllib/pom.xml
----------------------------------------------------------------------
diff --cc mllib/pom.xml
index 4cff5e3,f472082..a57bdde
--- a/mllib/pom.xml
+++ b/mllib/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
- <artifactId>spark-mllib_2.9.3</artifactId>
++ <artifactId>spark-mllib_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project ML Library</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -34,7 -34,7 +34,7 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
@@@ -48,12 -48,12 +48,12 @@@
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_2.10</artifactId>
- <artifactId>scalatest_2.9.3</artifactId>
++ <artifactId>scalatest_${scala-short.version}</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.scalacheck</groupId>
- <artifactId>scalacheck_2.10</artifactId>
- <artifactId>scalacheck_2.9.3</artifactId>
++ <artifactId>scalacheck_${scala-short.version}</artifactId>
<scope>test</scope>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/pom.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/project/SparkBuild.scala
----------------------------------------------------------------------
diff --cc project/SparkBuild.scala
index 19d3aa2,eb4b96e..2c63c0f
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@@ -153,17 -156,14 +153,18 @@@ object SparkBuild extends Build
*/
+
libraryDependencies ++= Seq(
- "org.eclipse.jetty" % "jetty-server" % "7.6.8.v20121106",
- "org.scalatest" %% "scalatest" % "1.9.1" % "test",
- "org.scalacheck" %% "scalacheck" % "1.10.0" % "test",
- "com.novocode" % "junit-interface" % "0.9" % "test",
- "org.easymock" % "easymock" % "3.1" % "test"
+ "io.netty" % "netty-all" % "4.0.0.CR1",
+ "org.eclipse.jetty" % "jetty-server" % "7.6.8.v20121106",
+ "org.scalatest" %% "scalatest" % "1.9.1" % "test",
+ "org.scalacheck" %% "scalacheck" % "1.10.0" % "test",
+ "com.novocode" % "junit-interface" % "0.9" % "test",
+ "org.easymock" % "easymock" % "3.1" % "test",
+ "commons-io" % "commons-io" % "2.4" % "test"
),
+
+ parallelExecution := false,
/* Workaround for issue #206 (fixed after SBT 0.11.0) */
watchTransitiveSources <<= Defaults.inDependencies[Task[Seq[File]]](watchSources.task,
const(std.TaskExtra.constant(Nil)), aggregate = true, includeRoot = true) apply { _.join.map(_.flatten) },
@@@ -192,35 -193,36 +194,34 @@@
),
libraryDependencies ++= Seq(
- "com.google.guava" % "guava" % "14.0.1",
- "com.google.code.findbugs" % "jsr305" % "1.3.9",
- "log4j" % "log4j" % "1.2.17",
- "org.slf4j" % "slf4j-api" % slf4jVersion,
- "org.slf4j" % "slf4j-log4j12" % slf4jVersion,
- "commons-daemon" % "commons-daemon" % "1.0.10", // workaround for bug HADOOP-9407
- "com.ning" % "compress-lzf" % "0.8.4",
- "org.xerial.snappy" % "snappy-java" % "1.0.5",
- "org.ow2.asm" % "asm" % "4.0",
- "com.google.protobuf" % "protobuf-java" % "2.4.1",
- "com.typesafe.akka" % "akka-actor" % "2.0.5" excludeAll(excludeNetty),
- "com.typesafe.akka" % "akka-remote" % "2.0.5" excludeAll(excludeNetty),
- "com.typesafe.akka" % "akka-slf4j" % "2.0.5" excludeAll(excludeNetty),
- "it.unimi.dsi" % "fastutil" % "6.4.4",
- "colt" % "colt" % "1.2.0",
- "net.liftweb" % "lift-json_2.9.2" % "2.5",
- "org.apache.mesos" % "mesos" % "0.13.0",
- "io.netty" % "netty-all" % "4.0.0.Beta2",
- "org.apache.derby" % "derby" % "10.4.2.0" % "test",
- "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib),
- "net.java.dev.jets3t" % "jets3t" % "0.7.1",
- "org.apache.avro" % "avro" % "1.7.4",
- "org.apache.avro" % "avro-ipc" % "1.7.4" excludeAll(excludeNetty),
- "com.codahale.metrics" % "metrics-core" % "3.0.0",
- "com.codahale.metrics" % "metrics-jvm" % "3.0.0",
- "com.codahale.metrics" % "metrics-json" % "3.0.0",
- "com.codahale.metrics" % "metrics-ganglia" % "3.0.0",
- "com.twitter" % "chill_2.9.3" % "0.3.1",
- "com.twitter" % "chill-java" % "0.3.1"
- )
+ "com.google.guava" % "guava" % "14.0.1",
+ "com.google.code.findbugs" % "jsr305" % "1.3.9",
+ "log4j" % "log4j" % "1.2.17",
+ "org.slf4j" % "slf4j-api" % slf4jVersion,
+ "org.slf4j" % "slf4j-log4j12" % slf4jVersion,
+ "com.ning" % "compress-lzf" % "0.8.4",
+ "org.xerial.snappy" % "snappy-java" % "1.0.5",
+ "commons-daemon" % "commons-daemon" % "1.0.10", // workaround for bug HADOOP-9407
+ "org.ow2.asm" % "asm" % "4.0",
+ "com.google.protobuf" % "protobuf-java" % "2.4.1",
- "de.javakaffee" % "kryo-serializers" % "0.22",
+ "com.typesafe.akka" %% "akka-remote" % "2.2.1" excludeAll(excludeNetty),
+ "com.typesafe.akka" %% "akka-slf4j" % "2.2.1" excludeAll(excludeNetty),
+ "net.liftweb" %% "lift-json" % "2.5.1" excludeAll(excludeNetty),
+ "it.unimi.dsi" % "fastutil" % "6.4.4",
+ "colt" % "colt" % "1.2.0",
+ "org.apache.mesos" % "mesos" % "0.13.0",
+ "net.java.dev.jets3t" % "jets3t" % "0.7.1",
+ "org.apache.derby" % "derby" % "10.4.2.0" % "test",
+ "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm),
+ "org.apache.avro" % "avro" % "1.7.4",
+ "org.apache.avro" % "avro-ipc" % "1.7.4" excludeAll(excludeNetty),
+ "com.codahale.metrics" % "metrics-core" % "3.0.0",
+ "com.codahale.metrics" % "metrics-jvm" % "3.0.0",
+ "com.codahale.metrics" % "metrics-json" % "3.0.0",
+ "com.codahale.metrics" % "metrics-ganglia" % "3.0.0",
+ "com.twitter" %% "chill" % "0.3.1",
+ "com.twitter" % "chill-java" % "0.3.1"
- )
++ )
)
def rootSettings = sharedSettings ++ Seq(
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/python/pyspark/rdd.py
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/repl-bin/pom.xml
----------------------------------------------------------------------
diff --cc repl-bin/pom.xml
index 05aadc7,f6bf94b..c983ea5
--- a/repl-bin/pom.xml
+++ b/repl-bin/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl-bin</artifactId>
- <artifactId>spark-repl-bin_2.9.3</artifactId>
++ <artifactId>spark-repl-bin_${scala-short.version}</artifactId>
<packaging>pom</packaging>
<name>Spark Project REPL binary packaging</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -40,18 -40,18 +40,18 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
- <artifactId>spark-bagel_2.9.3</artifactId>
++ <artifactId>spark-bagel_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
- <artifactId>spark-repl_2.9.3</artifactId>
++ <artifactId>spark-repl_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/repl/pom.xml
----------------------------------------------------------------------
diff --cc repl/pom.xml
index d4b1ea1,49d8662..ff66493
--- a/repl/pom.xml
+++ b/repl/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-repl</artifactId>
- <artifactId>spark-repl_2.9.3</artifactId>
++ <artifactId>spark-repl_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project REPL</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -39,18 -39,18 +39,18 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel</artifactId>
- <artifactId>spark-bagel_2.9.3</artifactId>
++ <artifactId>spark-bagel_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-mllib</artifactId>
- <artifactId>spark-mllib_2.9.3</artifactId>
++ <artifactId>spark-mllib_${scala-short.version}</artifactId>
<version>${project.version}</version>
<scope>runtime</scope>
</dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/streaming/pom.xml
----------------------------------------------------------------------
diff --cc streaming/pom.xml
index c74e76d,3b25fb4..3f2033f
--- a/streaming/pom.xml
+++ b/streaming/pom.xml
@@@ -26,7 -26,7 +26,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
- <artifactId>spark-streaming_2.9.3</artifactId>
++ <artifactId>spark-streaming_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project Streaming</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -42,7 -42,7 +42,7 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/tools/pom.xml
----------------------------------------------------------------------
diff --cc tools/pom.xml
index 2456457,f1c489b..db87b54
--- a/tools/pom.xml
+++ b/tools/pom.xml
@@@ -25,7 -25,7 +25,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-tools</artifactId>
- <artifactId>spark-tools_2.9.3</artifactId>
++ <artifactId>spark-tools_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project Tools</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -33,12 -33,12 +33,12 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming</artifactId>
- <artifactId>spark-streaming_2.9.3</artifactId>
++ <artifactId>spark-streaming_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/7be75682/yarn/pom.xml
----------------------------------------------------------------------
diff --cc yarn/pom.xml
index 7f852d3,3bc619d..7770cbb
--- a/yarn/pom.xml
+++ b/yarn/pom.xml
@@@ -25,7 -25,7 +25,7 @@@
</parent>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-yarn</artifactId>
- <artifactId>spark-yarn_2.9.3</artifactId>
++ <artifactId>spark-yarn_${scala-short.version}</artifactId>
<packaging>jar</packaging>
<name>Spark Project YARN Support</name>
<url>http://spark.incubator.apache.org/</url>
@@@ -33,7 -33,7 +33,7 @@@
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core</artifactId>
- <artifactId>spark-core_2.9.3</artifactId>
++ <artifactId>spark-core_${scala-short.version}</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
[20/50] git commit: SPARK-705: implement sortByKey() in PySpark
Posted by ma...@apache.org.
SPARK-705: implement sortByKey() in PySpark
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/fdbae41e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/fdbae41e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/fdbae41e
Branch: refs/heads/scala-2.10
Commit: fdbae41e88af0994e97ac8741f86547184a05de9
Parents: d585613
Author: Andre Schumacher <sc...@icsi.berkeley.edu>
Authored: Mon Oct 7 10:42:39 2013 -0700
Committer: Andre Schumacher <sc...@icsi.berkeley.edu>
Committed: Mon Oct 7 12:16:33 2013 -0700
----------------------------------------------------------------------
python/pyspark/rdd.py | 48 +++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 47 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/fdbae41e/python/pyspark/rdd.py
----------------------------------------------------------------------
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index 39c402b..7dfabb0 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -263,7 +263,53 @@ class RDD(object):
raise TypeError
return self.union(other)
- # TODO: sort
+ def sortByKey(self, ascending=True, numPartitions=None, keyfunc = lambda x: x):
+ """
+ Sorts this RDD, which is assumed to consist of (key, value) pairs.
+
+ >>> tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)]
+ >>> sc.parallelize(tmp).sortByKey(True, 2).collect()
+ [('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
+ >>> tmp2 = [('Mary', 1), ('had', 2), ('a', 3), ('little', 4), ('lamb', 5)]
+ >>> tmp2.extend([('whose', 6), ('fleece', 7), ('was', 8), ('white', 9)])
+ >>> sc.parallelize(tmp2).sortByKey(True, 3, keyfunc=lambda k: k.lower()).collect()
+ [('a', 3), ('fleece', 7), ('had', 2), ('lamb', 5), ('little', 4), ('Mary', 1), ('was', 8), ('white', 9), ('whose', 6)]
+ """
+ if numPartitions is None:
+ numPartitions = self.ctx.defaultParallelism
+
+ bounds = list()
+
+ # first compute the boundary of each part via sampling: we want to partition
+ # the key-space into bins such that the bins have roughly the same
+ # number of (key, value) pairs falling into them
+ if numPartitions > 1:
+ rddSize = self.count()
+ maxSampleSize = numPartitions * 20.0 # constant from Spark's RangePartitioner
+ fraction = min(maxSampleSize / max(rddSize, 1), 1.0)
+
+ samples = self.sample(False, fraction, 1).map(lambda (k, v): k).collect()
+ samples = sorted(samples, reverse=(not ascending), key=keyfunc)
+
+ # we have numPartitions many parts but one of the them has
+ # an implicit boundary
+ for i in range(0, numPartitions - 1):
+ index = (len(samples) - 1) * (i + 1) / numPartitions
+ bounds.append(samples[index])
+
+ def rangePartitionFunc(k):
+ p = 0
+ while p < len(bounds) and keyfunc(k) > bounds[p]:
+ p += 1
+ if ascending:
+ return p
+ else:
+ return numPartitions-1-p
+
+ def mapFunc(iterator):
+ yield sorted(iterator, reverse=(not ascending), key=lambda (k, v): keyfunc(k))
+
+ return self.partitionBy(numPartitions, partitionFunc=rangePartitionFunc).mapPartitions(mapFunc,preservesPartitioning=True).flatMap(lambda x: x, preservesPartitioning=True)
def glom(self):
"""
[19/50] git commit: Merge pull request #31 from sundeepn/branch-0.8
Posted by ma...@apache.org.
Merge pull request #31 from sundeepn/branch-0.8
Resolving package conflicts with hadoop 0.23.9
Hadoop 0.23.9 is having a package conflict with easymock's dependencies.
(cherry picked from commit 023e3fdf008b3194a36985a07923df9aaf64e520)
Signed-off-by: Reynold Xin <rx...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/213b70a2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/213b70a2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/213b70a2
Branch: refs/heads/scala-2.10
Commit: 213b70a2db5e92ada4b762bc39876c01a3530897
Parents: d585613
Author: Reynold Xin <rx...@apache.org>
Authored: Mon Oct 7 10:47:45 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Mon Oct 7 10:54:22 2013 -0700
----------------------------------------------------------------------
core/src/test/scala/org/apache/spark/ui/UISuite.scala | 7 ++++++-
project/SparkBuild.scala | 13 ++++++++-----
2 files changed, 14 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/213b70a2/core/src/test/scala/org/apache/spark/ui/UISuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/ui/UISuite.scala b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
index 07c9f23..8f0ec66 100644
--- a/core/src/test/scala/org/apache/spark/ui/UISuite.scala
+++ b/core/src/test/scala/org/apache/spark/ui/UISuite.scala
@@ -26,7 +26,12 @@ class UISuite extends FunSuite {
test("jetty port increases under contention") {
val startPort = 4040
val server = new Server(startPort)
- server.start()
+
+ Try { server.start() } match {
+ case Success(s) =>
+ case Failure(e) =>
+ // Either case server port is busy hence setup for test complete
+ }
val (jettyServer1, boundPort1) = JettyUtils.startJettyServer("localhost", startPort, Seq())
val (jettyServer2, boundPort2) = JettyUtils.startJettyServer("localhost", startPort, Seq())
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/213b70a2/project/SparkBuild.scala
----------------------------------------------------------------------
diff --git a/project/SparkBuild.scala b/project/SparkBuild.scala
index cdec616..eb4b96e 100644
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@ -156,6 +156,7 @@ object SparkBuild extends Build {
*/
+
libraryDependencies ++= Seq(
"org.eclipse.jetty" % "jetty-server" % "7.6.8.v20121106",
"org.scalatest" %% "scalatest" % "1.9.1" % "test",
@@ -178,6 +179,7 @@ object SparkBuild extends Build {
val slf4jVersion = "1.7.2"
+ val excludeCglib = ExclusionRule(organization = "org.sonatype.sisu.inject")
val excludeJackson = ExclusionRule(organization = "org.codehaus.jackson")
val excludeNetty = ExclusionRule(organization = "org.jboss.netty")
val excludeAsm = ExclusionRule(organization = "asm")
@@ -210,7 +212,7 @@ object SparkBuild extends Build {
"org.apache.mesos" % "mesos" % "0.13.0",
"io.netty" % "netty-all" % "4.0.0.Beta2",
"org.apache.derby" % "derby" % "10.4.2.0" % "test",
- "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm),
+ "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib),
"net.java.dev.jets3t" % "jets3t" % "0.7.1",
"org.apache.avro" % "avro" % "1.7.4",
"org.apache.avro" % "avro-ipc" % "1.7.4" excludeAll(excludeNetty),
@@ -248,6 +250,7 @@ object SparkBuild extends Build {
exclude("log4j","log4j")
exclude("org.apache.cassandra.deps", "avro")
excludeAll(excludeSnappy)
+ excludeAll(excludeCglib)
)
) ++ assemblySettings ++ extraAssemblySettings
@@ -290,10 +293,10 @@ object SparkBuild extends Build {
def yarnEnabledSettings = Seq(
libraryDependencies ++= Seq(
// Exclude rule required for all ?
- "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm),
- "org.apache.hadoop" % "hadoop-yarn-api" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm),
- "org.apache.hadoop" % "hadoop-yarn-common" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm),
- "org.apache.hadoop" % "hadoop-yarn-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm)
+ "org.apache.hadoop" % "hadoop-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib),
+ "org.apache.hadoop" % "hadoop-yarn-api" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib),
+ "org.apache.hadoop" % "hadoop-yarn-common" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib),
+ "org.apache.hadoop" % "hadoop-yarn-client" % hadoopVersion excludeAll(excludeJackson, excludeNetty, excludeAsm, excludeCglib)
)
)
[32/50] git commit: Merge pull request #43 from mateiz/kryo-fix
Posted by ma...@apache.org.
Merge pull request #43 from mateiz/kryo-fix
Don't allocate Kryo buffers unless needed
I noticed that the Kryo serializer could be slower than the Java one by 2-3x on small shuffles because it spend a lot of time initializing Kryo Input and Output objects. This is because our default buffer size for them is very large. Since the serializer is often used on streams, I made the initialization lazy for that, and used a smaller buffer (auto-managed by Kryo) for input.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/e67d5b96
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/e67d5b96
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/e67d5b96
Branch: refs/heads/scala-2.10
Commit: e67d5b962a2adddc073cfc9c99be9012fbb69838
Parents: ea34c52 a8725bf
Author: Reynold Xin <rx...@apache.org>
Authored: Tue Oct 8 22:57:38 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Tue Oct 8 22:57:38 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/serializer/KryoSerializer.scala | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
[42/50] git commit: Style fixes
Posted by ma...@apache.org.
Style fixes
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/36966f65
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/36966f65
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/36966f65
Branch: refs/heads/scala-2.10
Commit: 36966f65df2947a78d32d731f6b004d015ff011e
Parents: 3f7e9b2
Author: Kay Ousterhout <ka...@gmail.com>
Authored: Wed Oct 9 15:36:34 2013 -0700
Committer: Kay Ousterhout <ka...@gmail.com>
Committed: Wed Oct 9 15:36:34 2013 -0700
----------------------------------------------------------------------
.../spark/scheduler/SparkListenerSuite.scala | 19 +++++++++++++------
1 file changed, 13 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/36966f65/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index 6e80262..0d8742c 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -54,16 +54,23 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
assert(sc.dagScheduler.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS))
listener.stageInfos.size should be (4)
- listener.stageInfos.foreach {stageInfo =>
- //small test, so some tasks might take less than 1 millisecond, but average should be greater than 1 ms
+ listener.stageInfos.foreach { stageInfo =>
+ /* small test, so some tasks might take less than 1 millisecond, but average should be greater
+ * than 0 ms. */
checkNonZeroAvg(stageInfo.taskInfos.map{_._1.duration}, stageInfo + " duration")
- checkNonZeroAvg(stageInfo.taskInfos.map{_._2.executorRunTime.toLong}, stageInfo + " executorRunTime")
- checkNonZeroAvg(stageInfo.taskInfos.map{_._2.executorDeserializeTime.toLong}, stageInfo + " executorDeserializeTime")
+ checkNonZeroAvg(
+ stageInfo.taskInfos.map{_._2.executorRunTime.toLong},
+ stageInfo + " executorRunTime")
+ checkNonZeroAvg(
+ stageInfo.taskInfos.map{_._2.executorDeserializeTime.toLong},
+ stageInfo + " executorDeserializeTime")
if (stageInfo.stage.rdd.name == d4.name) {
- checkNonZeroAvg(stageInfo.taskInfos.map{_._2.shuffleReadMetrics.get.fetchWaitTime}, stageInfo + " fetchWaitTime")
+ checkNonZeroAvg(
+ stageInfo.taskInfos.map{_._2.shuffleReadMetrics.get.fetchWaitTime},
+ stageInfo + " fetchWaitTime")
}
- stageInfo.taskInfos.foreach{case (taskInfo, taskMetrics) =>
+ stageInfo.taskInfos.foreach { case (taskInfo, taskMetrics) =>
taskMetrics.resultSize should be > (0l)
if (isStage(stageInfo, Set(d2.name, d3.name), Set(d4.name))) {
taskMetrics.shuffleWriteMetrics should be ('defined)
[36/50] git commit: Merge pull request #38 from
AndreSchumacher/pyspark_sorting
Posted by ma...@apache.org.
Merge pull request #38 from AndreSchumacher/pyspark_sorting
SPARK-705: implement sortByKey() in PySpark
This PR contains the implementation of a RangePartitioner in Python and uses its partition ID's to get a global sort in PySpark.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/b4fa11f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/b4fa11f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/b4fa11f6
Branch: refs/heads/scala-2.10
Commit: b4fa11f6c96ee37ecd30231c1e22630055f52115
Parents: 19d445d fdbae41
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Wed Oct 9 11:59:47 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Wed Oct 9 11:59:47 2013 -0700
----------------------------------------------------------------------
python/pyspark/rdd.py | 48 +++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 47 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
[24/50] git commit: Don't allocate Kryo buffers unless needed
Posted by ma...@apache.org.
Don't allocate Kryo buffers unless needed
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a8725bf8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a8725bf8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a8725bf8
Branch: refs/heads/scala-2.10
Commit: a8725bf8f82ffea215afe7dd6c9ea1df36618e5b
Parents: 4a25b11
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Mon Oct 7 15:31:30 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Mon Oct 7 19:16:35 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/serializer/KryoSerializer.scala | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a8725bf8/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
index 24ef204..6c500ba 100644
--- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -38,8 +38,6 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
def newKryoOutput() = new KryoOutput(bufferSize)
- def newKryoInput() = new KryoInput(bufferSize)
-
def newKryo(): Kryo = {
val instantiator = new ScalaKryoInstantiator
val kryo = instantiator.newKryo()
@@ -118,8 +116,10 @@ class KryoDeserializationStream(kryo: Kryo, inStream: InputStream) extends Deser
private[spark] class KryoSerializerInstance(ks: KryoSerializer) extends SerializerInstance {
val kryo = ks.newKryo()
- val output = ks.newKryoOutput()
- val input = ks.newKryoInput()
+
+ // Make these lazy vals to avoid creating a buffer unless we use them
+ lazy val output = ks.newKryoOutput()
+ lazy val input = new KryoInput()
def serialize[T](t: T): ByteBuffer = {
output.clear()
[02/50] git commit: Style fix using 'if' rather than 'match' on
boolean
Posted by ma...@apache.org.
Style fix using 'if' rather than 'match' on boolean
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c6ceaeae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c6ceaeae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c6ceaeae
Branch: refs/heads/scala-2.10
Commit: c6ceaeae50c7b5d69bb9897af32f537bfbde152d
Parents: 6a7836c
Author: Nick Pentreath <ni...@gmail.com>
Authored: Fri Oct 4 13:52:53 2013 +0200
Committer: Nick Pentreath <ni...@gmail.com>
Committed: Fri Oct 4 13:52:53 2013 +0200
----------------------------------------------------------------------
.../apache/spark/mllib/recommendation/ALS.scala | 27 ++++++++++----------
1 file changed, 13 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c6ceaeae/mllib/src/main/scala/org/apache/spark/mllib/recommendation/ALS.scala
----------------------------------------------------------------------
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/recommendation/ALS.scala b/mllib/src/main/scala/org/apache/spark/mllib/recommendation/ALS.scala
index 5935d27..36853ac 100644
--- a/mllib/src/main/scala/org/apache/spark/mllib/recommendation/ALS.scala
+++ b/mllib/src/main/scala/org/apache/spark/mllib/recommendation/ALS.scala
@@ -207,20 +207,19 @@ class ALS private (var numBlocks: Int, var rank: Int, var iterations: Int, var l
* @return Option[YtY] - whose value is only used in the implicit preference model
*/
def computeYtY(factors: RDD[(Int, Array[Array[Double]])]) = {
- implicitPrefs match {
- case true => {
- Option(
- factors.flatMapValues{ case factorArray =>
- factorArray.map{ vector =>
- val x = new DoubleMatrix(vector)
- x.mmul(x.transpose())
- }
- }.reduceByKeyLocally((a, b) => a.addi(b))
- .values
- .reduce((a, b) => a.addi(b))
- )
- }
- case false => None
+ if (implicitPrefs) {
+ Option(
+ factors.flatMapValues{ case factorArray =>
+ factorArray.map{ vector =>
+ val x = new DoubleMatrix(vector)
+ x.mmul(x.transpose())
+ }
+ }.reduceByKeyLocally((a, b) => a.addi(b))
+ .values
+ .reduce((a, b) => a.addi(b))
+ )
+ } else {
+ None
}
}
[29/50] git commit: Revert "SPARK-900 Use coarser grained naming for
metrics"
Posted by ma...@apache.org.
Revert "SPARK-900 Use coarser grained naming for metrics"
This reverts commit 4b68be5f3c0a251453c184b233b3ca490812dafd.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/188abbf8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/188abbf8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/188abbf8
Branch: refs/heads/scala-2.10
Commit: 188abbf8f1765d3f52eff8b852309352630ab8fb
Parents: a2af6b5
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 17:45:14 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 17:45:14 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 5 +-
.../spark/deploy/master/MasterSource.scala | 7 +--
.../spark/deploy/worker/WorkerSource.scala | 11 ++--
.../apache/spark/executor/ExecutorSource.scala | 19 +++---
.../spark/scheduler/DAGSchedulerSource.scala | 11 ++--
.../spark/storage/BlockManagerSource.scala | 9 ++-
.../apache/spark/util/NamingConventions.scala | 62 --------------------
7 files changed, 28 insertions(+), 96 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index c72322e..5a24042 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -20,7 +20,6 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
class ApplicationSource(val application: ApplicationInfo) extends Source {
val metricRegistry = new MetricRegistry()
@@ -31,11 +30,11 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: String = application.state.toString
})
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("runtime", "ms")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("runtime_ms"), new Gauge[Long] {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("cores", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores", "number"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index de39398..23d1cb7 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -20,24 +20,23 @@ package org.apache.spark.deploy.master
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
private[spark] class MasterSource(val master: Master) extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("workers","number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers","number"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("apps", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps", "number"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("waitingApps", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps", "number"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index fc4f4ae..df269fd 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -20,33 +20,32 @@ package org.apache.spark.deploy.worker
import com.codahale.metrics.{Gauge, MetricRegistry}
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("executors", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors", "number"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresUsed", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed", "number"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memUsed", "MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("coresFree", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree", "number"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name(NamingConventions.makeMetricName("memFree", "MBytes")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memFree", "MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 6cbd154..18c9dc1 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -26,7 +26,6 @@ import org.apache.hadoop.fs.LocalFileSystem
import scala.collection.JavaConversions._
import org.apache.spark.metrics.source.Source
-import org.apache.spark.util.NamingConventions
class ExecutorSource(val executor: Executor, executorId: String) extends Source {
private def fileStats(scheme: String) : Option[FileSystem.Statistics] =
@@ -44,31 +43,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("activeTask", "count")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTask", "count"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("completeTask", "count")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTask", "count"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("currentPool", "size")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "currentPool", "size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getPoolSize()
})
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", NamingConventions.makeMetricName("maxPool", "size")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "maxPool", "size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getMaximumPoolSize()
})
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "bytes"), _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "bytes"), _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("read", "ops"), _.getReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("largeRead", "ops"), _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, NamingConventions.makeMetricName("write", "ops"), _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, "bytesRead", _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, "bytesWritten", _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, "readOps", _.getReadOps(), 0)
+ registerFileSystemStat(scheme, "largeReadOps", _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, "writeOps", _.getWriteOps(), 0)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 9e90a08..446d490 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -21,30 +21,29 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
-import org.apache.spark.util.NamingConventions
private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: SparkContext)
extends Source {
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("failedStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("runningStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", NamingConventions.makeMetricName("waitingStages", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("allJobs", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", NamingConventions.makeMetricName("activeJobs", "number")), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs", "number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index 4312250..acc3951 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -21,7 +21,6 @@ import com.codahale.metrics.{Gauge,MetricRegistry}
import org.apache.spark.metrics.source.Source
import org.apache.spark.SparkContext
-import org.apache.spark.util.NamingConventions
private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: SparkContext)
@@ -29,7 +28,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("maxMem", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "maxMem", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -37,7 +36,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("remainingMem", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "remainingMem", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -45,7 +44,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", NamingConventions.makeMetricName("memUsed", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "memUsed", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -54,7 +53,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", NamingConventions.makeMetricName("diskSpaceUsed", "MBytes")), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed", "MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/188abbf8/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/NamingConventions.scala b/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
deleted file mode 100644
index 7263361..0000000
--- a/core/src/main/scala/org/apache/spark/util/NamingConventions.scala
+++ /dev/null
@@ -1,62 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.spark.util
-
-/**
- * all utilities related to naming conventions
- */
-private[spark] object NamingConventions {
-
- /**
- * Lower camelCase which convert the phrases into camelCase style with the first letter lowercase
- */
- def lowerCamelCaseConversion(phrases: Seq[String]): Seq[String] = {
- var first = true
-
- for (elem <- phrases) yield {
- if (first) {
- first = false
- elem
- }
- else {
- elem.capitalize
- }
- }
- }
-
- /**
- * The standard camelCase style
- */
- def camelCaseConversion(phrases: Seq[String]): Seq[String] = {
- phrases.map(_.capitalize)
- }
-
- def noConversion = { x: Seq[String] => x }
-
- /**
- * Concatenate the words using certain naming style.
- * The default style is lowerCamelCase with empty connector.
- */
- def makeIdentifier(phrases: Seq[String], namingConversion: (Seq[String]) => Seq[String] = lowerCamelCaseConversion) (implicit connector: String = "" ): String = {
- namingConversion(phrases.filter(_.nonEmpty)).mkString(connector)
- }
-
- def makeMetricName(phrases: String*): String = {
- makeIdentifier(phrases, noConversion)("_")
- }
-}
\ No newline at end of file
[43/50] git commit: Fix Chill serialization of Range objects,
which used to write out each element,
and register user and Spark classes before Chill's serializers to let them
override Chill's behavior in general.
Posted by ma...@apache.org.
Fix Chill serialization of Range objects, which used to write out each
element, and register user and Spark classes before Chill's serializers
to let them override Chill's behavior in general.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c84c2052
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c84c2052
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c84c2052
Branch: refs/heads/scala-2.10
Commit: c84c2052898cb055012b8a6da00b8990cd8302c4
Parents: 7b3ae04
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Wed Oct 9 14:38:38 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Wed Oct 9 16:23:40 2013 -0700
----------------------------------------------------------------------
.../spark/serializer/KryoSerializer.scala | 14 ++++++++++---
.../spark/serializer/KryoSerializerSuite.scala | 21 ++++++++++++++++++++
2 files changed, 32 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84c2052/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
index 6c500ba..e936b1c 100644
--- a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
+++ b/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala
@@ -23,7 +23,7 @@ import java.io.{EOFException, InputStream, OutputStream}
import com.esotericsoftware.kryo.serializers.{JavaSerializer => KryoJavaSerializer}
import com.esotericsoftware.kryo.{KryoException, Kryo}
import com.esotericsoftware.kryo.io.{Input => KryoInput, Output => KryoOutput}
-import com.twitter.chill.ScalaKryoInstantiator
+import com.twitter.chill.{EmptyScalaKryoInstantiator, AllScalaRegistrar}
import org.apache.spark.{SerializableWritable, Logging}
import org.apache.spark.storage.{GetBlock, GotBlock, PutBlock, StorageLevel}
@@ -39,7 +39,7 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
def newKryoOutput() = new KryoOutput(bufferSize)
def newKryo(): Kryo = {
- val instantiator = new ScalaKryoInstantiator
+ val instantiator = new EmptyScalaKryoInstantiator
val kryo = instantiator.newKryo()
val classLoader = Thread.currentThread.getContextClassLoader
@@ -49,7 +49,11 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
StorageLevel.MEMORY_ONLY,
PutBlock("1", ByteBuffer.allocate(1), StorageLevel.MEMORY_ONLY),
GotBlock("1", ByteBuffer.allocate(1)),
- GetBlock("1")
+ GetBlock("1"),
+ 1 to 10,
+ 1 until 10,
+ 1L to 10L,
+ 1L until 10L
)
for (obj <- toRegister) kryo.register(obj.getClass)
@@ -69,6 +73,10 @@ class KryoSerializer extends org.apache.spark.serializer.Serializer with Logging
case _: Exception => println("Failed to register spark.kryo.registrator")
}
+ // Register Chill's classes; we do this after our ranges and the user's own classes to let
+ // our code override the generic serialziers in Chill for things like Seq
+ new AllScalaRegistrar().apply(kryo)
+
kryo.setClassLoader(classLoader)
// Allow disabling Kryo reference tracking if user knows their object graphs don't have loops
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c84c2052/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
index 0164dda..c016c51 100644
--- a/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/serializer/KryoSerializerSuite.scala
@@ -103,6 +103,27 @@ class KryoSerializerSuite extends FunSuite with SharedSparkContext {
check(List(mutable.HashMap("one" -> 1, "two" -> 2),mutable.HashMap(1->"one",2->"two",3->"three")))
}
+ test("ranges") {
+ val ser = (new KryoSerializer).newInstance()
+ def check[T](t: T) {
+ assert(ser.deserialize[T](ser.serialize(t)) === t)
+ // Check that very long ranges don't get written one element at a time
+ assert(ser.serialize(t).limit < 100)
+ }
+ check(1 to 1000000)
+ check(1 to 1000000 by 2)
+ check(1 until 1000000)
+ check(1 until 1000000 by 2)
+ check(1L to 1000000L)
+ check(1L to 1000000L by 2L)
+ check(1L until 1000000L)
+ check(1L until 1000000L by 2L)
+ check(1.0 to 1000000.0 by 1.0)
+ check(1.0 to 1000000.0 by 2.0)
+ check(1.0 until 1000000.0 by 1.0)
+ check(1.0 until 1000000.0 by 2.0)
+ }
+
test("custom registrator") {
System.setProperty("spark.kryo.registrator", classOf[MyRegistrator].getName)
[30/50] git commit: create metrics name manually.
Posted by ma...@apache.org.
create metrics name manually.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/22bed59d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/22bed59d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/22bed59d
Branch: refs/heads/scala-2.10
Commit: 22bed59d2de15d89ab5a778bea7439711e3cb13b
Parents: 188abbf
Author: Grace Huang <ji...@intel.com>
Authored: Tue Oct 8 18:01:11 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Tue Oct 8 18:01:11 2013 +0800
----------------------------------------------------------------------
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../apache/spark/deploy/worker/WorkerSource.scala | 10 +++++-----
.../apache/spark/executor/ExecutorSource.scala | 18 +++++++++---------
.../spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
.../apache/spark/storage/BlockManagerSource.scala | 8 ++++----
6 files changed, 27 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index 5a24042..01dbb3d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -34,7 +34,7 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name("cores", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores_number"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index 23d1cb7..a0f6fc4 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -26,17 +26,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name("workers","number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers_number"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name("apps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps_number"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name("waitingApps", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps_number"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index df269fd..5c2d3ee 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -25,27 +25,27 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name("executors", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors_number"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name("coresUsed", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed_number"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
// Gauge for memory used of this worker
- metricRegistry.register(MetricRegistry.name("memUsed", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memUsed_MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryUsed
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name("coresFree", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree_number"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
// Gauge for memory free of this worker
- metricRegistry.register(MetricRegistry.name("memFree", "MBytes"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("memFree_MBytes"), new Gauge[Int] {
override def getValue: Int = worker.memoryFree
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 18c9dc1..2123904 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -43,31 +43,31 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", "activeTask", "count"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTask_count"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", "completeTask", "count"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTask_count"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
// Gauge for executor thread pool's current number of threads
- metricRegistry.register(MetricRegistry.name("threadpool", "currentPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "currentPool_size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getPoolSize()
})
// Gauge got executor thread pool's largest number of threads that have ever simultaneously been in th pool
- metricRegistry.register(MetricRegistry.name("threadpool", "maxPool", "size"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "maxPool_size"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getMaximumPoolSize()
})
// Gauge for file system stats of this executor
for (scheme <- Array("hdfs", "file")) {
- registerFileSystemStat(scheme, "bytesRead", _.getBytesRead(), 0L)
- registerFileSystemStat(scheme, "bytesWritten", _.getBytesWritten(), 0L)
- registerFileSystemStat(scheme, "readOps", _.getReadOps(), 0)
- registerFileSystemStat(scheme, "largeReadOps", _.getLargeReadOps(), 0)
- registerFileSystemStat(scheme, "writeOps", _.getWriteOps(), 0)
+ registerFileSystemStat(scheme, "read_bytes", _.getBytesRead(), 0L)
+ registerFileSystemStat(scheme, "write_bytes", _.getBytesWritten(), 0L)
+ registerFileSystemStat(scheme, "read_ops", _.getReadOps(), 0)
+ registerFileSystemStat(scheme, "largeRead_ops", _.getLargeReadOps(), 0)
+ registerFileSystemStat(scheme, "write_ops", _.getWriteOps(), 0)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 446d490..474f3e8 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -27,23 +27,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", "failedStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", "runningStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", "waitingStages", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", "allJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", "activeJobs", "number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs_number"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/22bed59d/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
index acc3951..e5068d5 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManagerSource.scala
@@ -28,7 +28,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.BlockManager".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("memory", "maxMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "maxMem_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -36,7 +36,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "remainingMem", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "remainingMem_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val remainingMem = storageStatusList.map(_.memRemaining).reduce(_ + _)
@@ -44,7 +44,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("memory", "memUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("memory", "memUsed_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val maxMem = storageStatusList.map(_.maxMem).reduce(_ + _)
@@ -53,7 +53,7 @@ private[spark] class BlockManagerSource(val blockManager: BlockManager, sc: Spar
}
})
- metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed", "MBytes"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("disk", "diskSpaceUsed_MBytes"), new Gauge[Long] {
override def getValue: Long = {
val storageStatusList = blockManager.master.getStorageStatus
val diskSpaceUsed = storageStatusList
[03/50] git commit: Adding implicit feedback ALS to MLlib user guide
Posted by ma...@apache.org.
Adding implicit feedback ALS to MLlib user guide
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/93b96b44
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/93b96b44
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/93b96b44
Branch: refs/heads/scala-2.10
Commit: 93b96b44d778716a4e76bdcf68d6a07694a06460
Parents: c6ceaea
Author: Nick Pentreath <ni...@gmail.com>
Authored: Fri Oct 4 14:39:44 2013 +0200
Committer: Nick Pentreath <ni...@gmail.com>
Committed: Fri Oct 4 14:39:44 2013 +0200
----------------------------------------------------------------------
docs/mllib-guide.md | 24 ++++++++++++++++++++----
1 file changed, 20 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/93b96b44/docs/mllib-guide.md
----------------------------------------------------------------------
diff --git a/docs/mllib-guide.md b/docs/mllib-guide.md
index f991d86..c1ff9c4 100644
--- a/docs/mllib-guide.md
+++ b/docs/mllib-guide.md
@@ -144,10 +144,9 @@ Available algorithms for clustering:
# Collaborative Filtering
-[Collaborative
-filtering](http://en.wikipedia.org/wiki/Recommender_system#Collaborative_filtering)
+[Collaborative filtering](http://en.wikipedia.org/wiki/Recommender_system#Collaborative_filtering)
is commonly used for recommender systems. These techniques aim to fill in the
-missing entries of a user-product association matrix. MLlib currently supports
+missing entries of a user-item association matrix. MLlib currently supports
model-based collaborative filtering, in which users and products are described
by a small set of latent factors that can be used to predict missing entries.
In particular, we implement the [alternating least squares
@@ -158,7 +157,24 @@ following parameters:
* *numBlocks* is the number of blacks used to parallelize computation (set to -1 to auto-configure).
* *rank* is the number of latent factors in our model.
* *iterations* is the number of iterations to run.
-* *lambda* specifies the regularization parameter in ALS.
+* *lambda* specifies the regularization parameter in ALS.
+* *implicitPrefs* specifies whether to use the *explicit feedback* ALS variant or one adapted for *implicit feedback* data
+* *alpha* is a parameter applicable to the implicit feedback variant of ALS that governs the *baseline* confidence in preference observations
+
+## Explicit vs Implicit Feedback
+
+The standard approach to matrix factorization based collaborative filtering treats
+the entries in the user-item matrix as *explicit* preferences given by the user to the item.
+
+It is common in many real-world use cases to only have access to *implicit feedback*
+(e.g. views, clicks, purchases, likes, shares etc.). The approach used in MLlib to deal with
+such data is taken from
+[Collaborative Filtering for Implicit Feedback Datasets](http://research.yahoo.com/pub/2433).
+Essentially instead of trying to model the matrix of ratings directly, this approach treats the data as
+a combination of binary preferences and *confidence values*. The ratings are then related
+to the level of confidence in observed user preferences, rather than explicit ratings given to items.
+The model then tries to find latent factors that can be used to predict the expected preference of a user
+for an item.
Available algorithms for collaborative filtering:
[22/50] git commit: Merge pull request #39 from pwendell/master
Posted by ma...@apache.org.
Merge pull request #39 from pwendell/master
Adding Shark 0.7.1 to EC2 scripts
This adds a newer version of Shark to the ec2 scripts. I've tested this for both Hadoop1 and Hadoop2 clusters.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/02f37ee8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/02f37ee8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/02f37ee8
Branch: refs/heads/scala-2.10
Commit: 02f37ee8530bcd5afefe403147bbf411464fa773
Parents: 213b70a 3745a18
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Oct 7 15:48:52 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Oct 7 15:48:52 2013 -0700
----------------------------------------------------------------------
ec2/spark_ec2.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
[05/50] git commit: Merge branch 'master' into scala-2.10
Posted by ma...@apache.org.
Merge branch 'master' into scala-2.10
Conflicts:
core/src/test/scala/org/apache/spark/DistributedSuite.scala
project/SparkBuild.scala
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/c810ee06
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/c810ee06
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/c810ee06
Branch: refs/heads/scala-2.10
Commit: c810ee0690d536873b1046541cdf1b492902a60e
Parents: 5829692 232765f
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Sat Oct 5 15:52:57 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Sat Oct 5 15:52:57 2013 +0530
----------------------------------------------------------------------
.../scala/org/apache/spark/SparkContext.scala | 2 +-
.../scala/org/apache/spark/TaskEndReason.scala | 8 +-
.../org/apache/spark/executor/Executor.scala | 26 +--
.../apache/spark/scheduler/DAGScheduler.scala | 5 +-
.../scala/org/apache/spark/scheduler/Pool.scala | 6 +-
.../apache/spark/scheduler/Schedulable.scala | 4 +-
.../org/apache/spark/scheduler/TaskResult.scala | 15 +-
.../apache/spark/scheduler/TaskSetManager.scala | 2 -
.../scheduler/cluster/ClusterScheduler.scala | 55 +++++--
.../cluster/ClusterTaskSetManager.scala | 162 +++++++++----------
.../scheduler/cluster/TaskResultGetter.scala | 124 ++++++++++++++
.../spark/scheduler/local/LocalScheduler.scala | 5 +-
.../scheduler/local/LocalTaskSetManager.scala | 21 ++-
.../org/apache/spark/storage/BlockManager.scala | 27 +++-
.../org/apache/spark/DistributedSuite.scala | 1 +
.../spark/scheduler/DAGSchedulerSuite.scala | 2 -
.../cluster/ClusterSchedulerSuite.scala | 15 +-
.../cluster/ClusterTaskSetManagerSuite.scala | 58 ++++++-
.../cluster/TaskResultGetterSuite.scala | 113 +++++++++++++
docs/running-on-yarn.md | 1 +
ec2/README | 2 +-
make-distribution.sh | 2 +-
project/SparkBuild.scala | 3 +
.../org/apache/spark/deploy/yarn/Client.scala | 2 +-
.../spark/deploy/yarn/ClientArguments.scala | 6 +
25 files changed, 510 insertions(+), 157 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/core/src/main/scala/org/apache/spark/executor/Executor.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/core/src/test/scala/org/apache/spark/DistributedSuite.scala
----------------------------------------------------------------------
diff --cc core/src/test/scala/org/apache/spark/DistributedSuite.scala
index c719a54,cd2bf9a..a31988a
--- a/core/src/test/scala/org/apache/spark/DistributedSuite.scala
+++ b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
@@@ -319,19 -319,6 +319,20 @@@ class DistributedSuite extends FunSuit
}
}
}
+
+ test("job should fail if TaskResult exceeds Akka frame size") {
+ // We must use local-cluster mode since results are returned differently
+ // when running under LocalScheduler:
+ sc = new SparkContext("local-cluster[1,1,512]", "test")
+ val akkaFrameSize =
+ sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
+ val rdd = sc.parallelize(Seq(1)).map{x => new Array[Byte](akkaFrameSize)}
+ val exception = intercept[SparkException] {
+ rdd.reduce((x, y) => x)
+ }
+ exception.getMessage should endWith("result exceeded Akka frame size")
+ }
++
}
object DistributedSuite {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/c810ee06/project/SparkBuild.scala
----------------------------------------------------------------------
diff --cc project/SparkBuild.scala
index 67d03f9,cdec616..19d3aa2
--- a/project/SparkBuild.scala
+++ b/project/SparkBuild.scala
@@@ -97,6 -97,12 +97,9 @@@ object SparkBuild extends Build
// Only allow one test at a time, even across projects, since they run in the same JVM
concurrentRestrictions in Global += Tags.limit(Tags.Test, 1),
+ // also check the local Maven repository ~/.m2
+ resolvers ++= Seq(Resolver.file("Local Maven Repo", file(Path.userHome + "/.m2/repository"))),
+
- // Shared between both core and streaming.
- resolvers ++= Seq("Akka Repository" at "http://repo.akka.io/releases/"),
-
// For Sonatype publishing
resolvers ++= Seq("sonatype-snapshots" at "https://oss.sonatype.org/content/repositories/snapshots",
"sonatype-staging" at "https://oss.sonatype.org/service/local/staging/deploy/maven2/"),
[11/50] git commit: Make HadoopRDD object Spark private.
Posted by ma...@apache.org.
Make HadoopRDD object Spark private.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/96929f28
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/96929f28
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/96929f28
Branch: refs/heads/scala-2.10
Commit: 96929f28bb9c929ca3309dbe99910097f5eb3c8c
Parents: b5e93c1
Author: Harvey Feng <ha...@databricks.com>
Authored: Sat Oct 5 17:14:19 2013 -0700
Committer: Harvey Feng <ha...@databricks.com>
Committed: Sat Oct 5 17:14:19 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/96929f28/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
index 728f3d1..51e5bb8 100644
--- a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
@@ -106,7 +106,7 @@ class HadoopRDD[K, V](
protected val jobConfCacheKey = "rdd_%d_job_conf".format(id)
- private val inputFormatCacheKey = "rdd_%d_input_format".format(id)
+ protected val inputFormatCacheKey = "rdd_%d_input_format".format(id)
// Returns a JobConf that will be used on slaves to obtain input splits for Hadoop reads.
protected def getJobConf(): JobConf = {
@@ -122,7 +122,7 @@ class HadoopRDD[K, V](
}
}
- def getInputFormat(conf: JobConf): InputFormat[K, V] = {
+ protected def getInputFormat(conf: JobConf): InputFormat[K, V] = {
if (HadoopRDD.containsCachedMetadata(inputFormatCacheKey)) {
return HadoopRDD.getCachedMetadata(inputFormatCacheKey).asInstanceOf[InputFormat[K, V]]
}
@@ -196,7 +196,7 @@ class HadoopRDD[K, V](
def getConf: Configuration = getJobConf()
}
-object HadoopRDD {
+private[spark] object HadoopRDD {
def getCachedMetadata(key: String) = SparkEnv.get.hadoop.hadoopJobMetadata.get(key)
def containsCachedMetadata(key: String) = SparkEnv.get.hadoop.hadoopJobMetadata.containsKey(key)
[40/50] git commit: Fix race condition in SparkListenerSuite (fixes
SPARK-908).
Posted by ma...@apache.org.
Fix race condition in SparkListenerSuite (fixes SPARK-908).
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a34a4e81
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a34a4e81
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a34a4e81
Branch: refs/heads/scala-2.10
Commit: a34a4e8174b5f285a327d7ff30ac9f3ff0db7689
Parents: 3218fa7
Author: Kay Ousterhout <ka...@gmail.com>
Authored: Wed Oct 9 15:07:53 2013 -0700
Committer: Kay Ousterhout <ka...@gmail.com>
Committed: Wed Oct 9 15:07:53 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/SparkContext.scala | 2 +-
.../org/apache/spark/scheduler/DAGScheduler.scala | 2 +-
.../apache/spark/scheduler/SparkListenerBus.scala | 18 ++++++++++++++++++
.../spark/scheduler/SparkListenerSuite.scala | 15 +++++----------
4 files changed, 25 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a34a4e81/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index febcf9c..ff45e76 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -231,7 +231,7 @@ class SparkContext(
}
taskScheduler.start()
- @volatile private var dagScheduler = new DAGScheduler(taskScheduler)
+ @volatile private[spark] var dagScheduler = new DAGScheduler(taskScheduler)
dagScheduler.start()
ui.start()
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a34a4e81/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 4053b91..5c40f50 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -114,7 +114,7 @@ class DAGScheduler(
private[spark] val stageToInfos = new TimeStampedHashMap[Stage, StageInfo]
- private val listenerBus = new SparkListenerBus()
+ private[spark] val listenerBus = new SparkListenerBus()
// Contains the locations that each RDD's partitions are cached on
private val cacheLocs = new HashMap[Int, Array[Seq[TaskLocation]]]
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a34a4e81/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
index a65e1ec..8283c4b 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala
@@ -70,5 +70,23 @@ private[spark] class SparkListenerBus() extends Logging {
queueFullErrorMessageLogged = true
}
}
+
+ /** Waits until there are no more events in the queue, or until the specified time has elapsed.
+ *
+ * Used for testing only. Returns true if the queue has emptied and false is the specified time
+ * elapsed before the queue emptied.
+ */
+ def waitUntilEmpty(timeoutMillis: Int): Boolean = {
+ val finishTime = System.currentTimeMillis + timeoutMillis
+ while (!eventQueue.isEmpty()) {
+ if (System.currentTimeMillis > finishTime) {
+ return false
+ }
+ /* Sleep rather than using wait/notify, because this is used only for testing and wait/notify
+ * add overhead in the general case. */
+ Thread.sleep(10)
+ }
+ return true
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/a34a4e81/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
index 41a161e..6e80262 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/SparkListenerSuite.scala
@@ -23,15 +23,9 @@ import scala.collection.mutable
import org.scalatest.matchers.ShouldMatchers
import org.apache.spark.SparkContext._
-/**
- *
- */
-
class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatchers {
- // TODO: This test has a race condition since the DAGScheduler now reports results
- // asynchronously. It needs to be updated for that patch.
- ignore("local metrics") {
+ test("local metrics") {
sc = new SparkContext("local[4]", "test")
val listener = new SaveStageInfo
sc.addSparkListener(listener)
@@ -45,7 +39,8 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
val d = sc.parallelize(1 to 1e4.toInt, 64).map{i => w(i)}
d.count
- Thread.sleep(1000)
+ val WAIT_TIMEOUT_MILLIS = 10000
+ assert(sc.dagScheduler.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS))
listener.stageInfos.size should be (1)
val d2 = d.map{i => w(i) -> i * 2}.setName("shuffle input 1")
@@ -57,7 +52,7 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
d4.collectAsMap
- Thread.sleep(1000)
+ assert(sc.dagScheduler.listenerBus.waitUntilEmpty(WAIT_TIMEOUT_MILLIS))
listener.stageInfos.size should be (4)
listener.stageInfos.foreach {stageInfo =>
//small test, so some tasks might take less than 1 millisecond, but average should be greater than 1 ms
@@ -68,7 +63,7 @@ class SparkListenerSuite extends FunSuite with LocalSparkContext with ShouldMatc
checkNonZeroAvg(stageInfo.taskInfos.map{_._2.shuffleReadMetrics.get.fetchWaitTime}, stageInfo + " fetchWaitTime")
}
- stageInfo.taskInfos.foreach{case (taskInfo, taskMetrics) =>
+ stageInfo.taskInfos.foreach{case (taskInfo, taskMetrics) =>
taskMetrics.resultSize should be > (0l)
if (isStage(stageInfo, Set(d2.name, d3.name), Set(d4.name))) {
taskMetrics.shuffleWriteMetrics should be ('defined)
[18/50] git commit: Merge branch 'master' into implicit-als
Posted by ma...@apache.org.
Merge branch 'master' into implicit-als
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/a5e58b8f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/a5e58b8f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/a5e58b8f
Branch: refs/heads/scala-2.10
Commit: a5e58b8f980fe49d02dce83e366f0ea0cf070d76
Parents: b0f5f4d d585613
Author: Nick Pentreath <ni...@gmail.com>
Authored: Mon Oct 7 11:46:17 2013 +0200
Committer: Nick Pentreath <ni...@gmail.com>
Committed: Mon Oct 7 11:46:17 2013 +0200
----------------------------------------------------------------------
assembly/pom.xml | 18 +--
bagel/pom.xml | 8 +-
core/pom.xml | 15 +-
.../scala/org/apache/spark/CacheManager.scala | 4 +-
.../scala/org/apache/spark/SparkContext.scala | 39 ++++--
.../spark/api/python/PythonPartitioner.scala | 10 +-
.../org/apache/spark/api/python/PythonRDD.scala | 6 +-
.../org/apache/spark/deploy/JsonProtocol.scala | 3 +-
.../apache/spark/deploy/SparkHadoopUtil.scala | 12 +-
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 140 ++++++++++++++++---
.../scala/org/apache/spark/util/Utils.scala | 13 ++
.../spark/scheduler/SparkListenerSuite.scala | 6 +-
.../org/apache/spark/util/UtilsSuite.scala | 11 ++
docs/_layouts/global.html | 4 +-
ec2/spark_ec2.py | 4 +-
examples/pom.xml | 30 ++--
mllib/pom.xml | 8 +-
pom.xml | 7 +-
python/pyspark/rdd.py | 10 +-
python/pyspark/serializers.py | 4 +
repl-bin/pom.xml | 10 +-
repl/pom.xml | 20 +--
streaming/pom.xml | 9 +-
tools/pom.xml | 8 +-
yarn/pom.xml | 6 +-
25 files changed, 293 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
[16/50] git commit: Merge pull request #37 from pwendell/merge-0.8
Posted by ma...@apache.org.
Merge pull request #37 from pwendell/merge-0.8
merge in remaining changes from `branch-0.8`
This merges in the following changes from `branch-0.8`:
- The scala version is included in the published maven artifact names
- A unit tests which had non-deterministic failures is ignored (see SPARK-908)
- A minor documentation change shows the short version instead of the full version
- Moving the kafka jar to be "provided"
- Changing the default spark ec2 version.
- Some spacing changes caused by Maven's release plugin
Note that I've squashed this into a single commit rather than pull in the branch-0.8 history. There are a bunch of release/revert commits there that make the history super ugly.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/d585613e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/d585613e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/d585613e
Branch: refs/heads/scala-2.10
Commit: d585613ee22ed9292a69e35cedb0d5965f4c906f
Parents: 4a25b11 aa9fb84
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sat Oct 5 22:57:05 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sat Oct 5 22:57:05 2013 -0700
----------------------------------------------------------------------
assembly/pom.xml | 18 ++++++------
bagel/pom.xml | 8 +++---
core/pom.xml | 15 +++++-----
.../spark/scheduler/SparkListenerSuite.scala | 6 +++-
docs/_layouts/global.html | 4 +--
examples/pom.xml | 30 ++++++++++++++------
mllib/pom.xml | 8 +++---
pom.xml | 7 +++--
repl-bin/pom.xml | 10 +++----
repl/pom.xml | 20 ++++++-------
streaming/pom.xml | 9 +++---
tools/pom.xml | 8 +++---
yarn/pom.xml | 6 ++--
13 files changed, 84 insertions(+), 65 deletions(-)
----------------------------------------------------------------------
[37/50] git commit: Fix PySpark docs and an overly long line of code
after fdbae41e
Posted by ma...@apache.org.
Fix PySpark docs and an overly long line of code after fdbae41e
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/478b2b7e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/478b2b7e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/478b2b7e
Branch: refs/heads/scala-2.10
Commit: 478b2b7edcf42fa3e16f625d4b8676f2bb31f8dc
Parents: b4fa11f
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Wed Oct 9 12:08:04 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Wed Oct 9 12:08:04 2013 -0700
----------------------------------------------------------------------
docs/python-programming-guide.md | 2 +-
python/pyspark/rdd.py | 16 ++++++++--------
2 files changed, 9 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/478b2b7e/docs/python-programming-guide.md
----------------------------------------------------------------------
diff --git a/docs/python-programming-guide.md b/docs/python-programming-guide.md
index f67a1cc..6c2336a 100644
--- a/docs/python-programming-guide.md
+++ b/docs/python-programming-guide.md
@@ -16,7 +16,7 @@ This guide will show how to use the Spark features described there in Python.
There are a few key differences between the Python and Scala APIs:
* Python is dynamically typed, so RDDs can hold objects of multiple types.
-* PySpark does not yet support a few API calls, such as `lookup`, `sort`, and non-text input files, though these will be added in future releases.
+* PySpark does not yet support a few API calls, such as `lookup` and non-text input files, though these will be added in future releases.
In PySpark, RDDs support the same methods as their Scala counterparts but take Python functions and return Python collection types.
Short functions can be passed to RDD methods using Python's [`lambda`](http://www.diveintopython.net/power_of_introspection/lambda_functions.html) syntax:
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/478b2b7e/python/pyspark/rdd.py
----------------------------------------------------------------------
diff --git a/python/pyspark/rdd.py b/python/pyspark/rdd.py
index 7dfabb0..7019fb8 100644
--- a/python/pyspark/rdd.py
+++ b/python/pyspark/rdd.py
@@ -117,8 +117,6 @@ class RDD(object):
else:
return None
- # TODO persist(self, storageLevel)
-
def map(self, f, preservesPartitioning=False):
"""
Return a new RDD containing the distinct elements in this RDD.
@@ -227,7 +225,7 @@ class RDD(object):
total = num
samples = self.sample(withReplacement, fraction, seed).collect()
-
+
# If the first sample didn't turn out large enough, keep trying to take samples;
# this shouldn't happen often because we use a big multiplier for their initial size.
# See: scala/spark/RDD.scala
@@ -288,7 +286,7 @@ class RDD(object):
maxSampleSize = numPartitions * 20.0 # constant from Spark's RangePartitioner
fraction = min(maxSampleSize / max(rddSize, 1), 1.0)
- samples = self.sample(False, fraction, 1).map(lambda (k, v): k).collect()
+ samples = self.sample(False, fraction, 1).map(lambda (k, v): k).collect()
samples = sorted(samples, reverse=(not ascending), key=keyfunc)
# we have numPartitions many parts but one of the them has
@@ -309,7 +307,9 @@ class RDD(object):
def mapFunc(iterator):
yield sorted(iterator, reverse=(not ascending), key=lambda (k, v): keyfunc(k))
- return self.partitionBy(numPartitions, partitionFunc=rangePartitionFunc).mapPartitions(mapFunc,preservesPartitioning=True).flatMap(lambda x: x, preservesPartitioning=True)
+ return (self.partitionBy(numPartitions, partitionFunc=rangePartitionFunc)
+ .mapPartitions(mapFunc,preservesPartitioning=True)
+ .flatMap(lambda x: x, preservesPartitioning=True))
def glom(self):
"""
@@ -471,7 +471,7 @@ class RDD(object):
3
"""
return self.mapPartitions(lambda i: [sum(1 for _ in i)]).sum()
-
+
def stats(self):
"""
Return a L{StatCounter} object that captures the mean, variance
@@ -508,7 +508,7 @@ class RDD(object):
0.816...
"""
return self.stats().stdev()
-
+
def sampleStdev(self):
"""
Compute the sample standard deviation of this RDD's elements (which corrects for bias in
@@ -878,7 +878,7 @@ class RDD(object):
>>> y = sc.parallelize([("a", 3), ("c", None)])
>>> sorted(x.subtractByKey(y).collect())
[('b', 4), ('b', 5)]
- """
+ """
filter_func = lambda (key, vals): len(vals[0]) > 0 and len(vals[1]) == 0
map_func = lambda (key, vals): [(key, val) for val in vals[0]]
return self.cogroup(other, numPartitions).filter(filter_func).flatMap(map_func)
[39/50] git commit: Merge pull request #46 from mateiz/py-sort-update
Posted by ma...@apache.org.
Merge pull request #46 from mateiz/py-sort-update
Fix PySpark docs and an overly long line of code after #38
Just noticed these after merging that commit (https://github.com/apache/incubator-spark/pull/38).
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/7827efc8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/7827efc8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/7827efc8
Branch: refs/heads/scala-2.10
Commit: 7827efc87b6652779cba37ccab9f87c7f31dd645
Parents: 7b3ae04 478b2b7
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Wed Oct 9 15:07:25 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Wed Oct 9 15:07:25 2013 -0700
----------------------------------------------------------------------
docs/python-programming-guide.md | 2 +-
python/pyspark/rdd.py | 16 ++++++++--------
2 files changed, 9 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
[38/50] git commit: Merge pull request #45 from pwendell/metrics_units
Posted by ma...@apache.org.
Merge pull request #45 from pwendell/metrics_units
Use standard abbreviation in metrics description (MBytes -> MB)
This is a small change - older commits are shown here because Github hasn't sync'ed yet with apache.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/7b3ae04e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/7b3ae04e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/7b3ae04e
Branch: refs/heads/scala-2.10
Commit: 7b3ae04ea7f95ee9a8f09d4615d1c9fbfc475af5
Parents: b4fa11f bd3bcc5
Author: Patrick Wendell <pw...@gmail.com>
Authored: Wed Oct 9 12:14:19 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Wed Oct 9 12:14:19 2013 -0700
----------------------------------------------------------------------
.../scala/org/apache/spark/deploy/worker/WorkerSource.scala | 4 ++--
.../scala/org/apache/spark/storage/BlockManagerSource.scala | 8 ++++----
2 files changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
[08/50] git commit: Merge pull request #27 from davidmccauley/master
Posted by ma...@apache.org.
Merge pull request #27 from davidmccauley/master
SPARK-920/921 - JSON endpoint updates
920 - Removal of duplicate scheme part of Spark URI, it was appearing as spark://spark//host:port in the JSON field.
JSON now delivered as:
url:spark://127.0.0.1:7077
921 - Adding the URL of the Main Application UI will allow custom interfaces (that use the JSON output) to redirect from the standalone UI.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/100222b0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/100222b0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/100222b0
Branch: refs/heads/scala-2.10
Commit: 100222b048a4957eeb4642d0cd3c54b2b468b0b4
Parents: 0864193 1577b37
Author: Matei Zaharia <ma...@eecs.berkeley.edu>
Authored: Sat Oct 5 13:38:59 2013 -0700
Committer: Matei Zaharia <ma...@eecs.berkeley.edu>
Committed: Sat Oct 5 13:38:59 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/deploy/JsonProtocol.scala | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
[10/50] git commit: Bumping EC2 default version in master to `0.8.0`.
Posted by ma...@apache.org.
Bumping EC2 default version in master to `0.8.0`.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/2484b846
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/2484b846
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/2484b846
Branch: refs/heads/scala-2.10
Commit: 2484b846788ca2ef5f0cebd625154a0333d50797
Parents: 232765f
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sat Oct 5 16:59:11 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Sat Oct 5 16:59:11 2013 -0700
----------------------------------------------------------------------
ec2/spark_ec2.py | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/2484b846/ec2/spark_ec2.py
----------------------------------------------------------------------
diff --git a/ec2/spark_ec2.py b/ec2/spark_ec2.py
index 1190ed4..7da9355 100755
--- a/ec2/spark_ec2.py
+++ b/ec2/spark_ec2.py
@@ -70,7 +70,7 @@ def parse_args():
"slaves across multiple (an additional $0.01/Gb for bandwidth" +
"between zones applies)")
parser.add_option("-a", "--ami", help="Amazon Machine Image ID to use")
- parser.add_option("-v", "--spark-version", default="0.7.3",
+ parser.add_option("-v", "--spark-version", default="0.8.0",
help="Version of Spark to use: 'X.Y.Z' or a specific git hash")
parser.add_option("--spark-git-repo",
default="https://github.com/mesos/spark",
@@ -155,7 +155,7 @@ def is_active(instance):
# Return correct versions of Spark and Shark, given the supplied Spark version
def get_spark_shark_version(opts):
- spark_shark_map = {"0.7.3": "0.7.0"}
+ spark_shark_map = {"0.7.3": "0.7.0", "0.8.0": "0.8.0"}
version = opts.spark_version.replace("v", "")
if version not in spark_shark_map:
print >> stderr, "Don't know about Spark version: %s" % version
[48/50] git commit: Changed message-frame-size to maximum-frame-size
as property. Removed a test accidentally added during merge.
Posted by ma...@apache.org.
Changed message-frame-size to maximum-frame-size as property.
Removed a test accidentally added during merge.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/34da58ae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/34da58ae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/34da58ae
Branch: refs/heads/scala-2.10
Commit: 34da58ae50cd97a1136d45484130addcf6ac8a33
Parents: 026ab75
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Thu Oct 10 14:03:56 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Thu Oct 10 15:13:44 2013 +0530
----------------------------------------------------------------------
.../scala/org/apache/spark/executor/Executor.scala | 2 +-
.../main/scala/org/apache/spark/util/AkkaUtils.scala | 4 ++--
.../test/scala/org/apache/spark/DistributedSuite.scala | 13 -------------
.../scheduler/cluster/TaskResultGetterSuite.scala | 4 ++--
4 files changed, 5 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/34da58ae/core/src/main/scala/org/apache/spark/executor/Executor.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index b4153f3..3800063 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/core/src/main/scala/org/apache/spark/executor/Executor.scala
@@ -105,7 +105,7 @@ private[spark] class Executor(
SparkEnv.set(env)
env.metricsSystem.registerSource(executorSource)
- private val akkaFrameSize = env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size")
+ private val akkaFrameSize = env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.maximum-frame-size")
// Start worker thread pool
val threadPool = new ThreadPoolExecutor(
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/34da58ae/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
index af1c36b..8daf50a 100644
--- a/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
+++ b/core/src/main/scala/org/apache/spark/util/AkkaUtils.scala
@@ -54,8 +54,8 @@ private[spark] object AkkaUtils {
akka.remote.netty.tcp.transport-class = "akka.remote.transport.netty.NettyTransport"
akka.remote.netty.tcp.hostname = "%s"
akka.remote.netty.tcp.port = %d
- akka.remote.netty.tcp.connection-timeout = %ds
- akka.remote.netty.tcp.message-frame-size = %d MiB
+ akka.remote.netty.tcp.connection-timeout = %d s
+ akka.remote.netty.tcp.maximum-frame-size = %dMiB
akka.remote.netty.tcp.execution-pool-size = %d
akka.actor.default-dispatcher.throughput = %d
akka.remote.log-remote-lifecycle-events = %s
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/34da58ae/core/src/test/scala/org/apache/spark/DistributedSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/DistributedSuite.scala b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
index 25b9c3e..988ab17 100644
--- a/core/src/test/scala/org/apache/spark/DistributedSuite.scala
+++ b/core/src/test/scala/org/apache/spark/DistributedSuite.scala
@@ -320,19 +320,6 @@ class DistributedSuite extends FunSuite with ShouldMatchers with BeforeAndAfter
}
}
- test("job should fail if TaskResult exceeds Akka frame size") {
- // We must use local-cluster mode since results are returned differently
- // when running under LocalScheduler:
- sc = new SparkContext("local-cluster[1,1,512]", "test")
- val akkaFrameSize =
- sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
- val rdd = sc.parallelize(Seq(1)).map{x => new Array[Byte](akkaFrameSize)}
- val exception = intercept[SparkException] {
- rdd.reduce((x, y) => x)
- }
- exception.getMessage should endWith("result exceeded Akka frame size")
- }
-
}
object DistributedSuite {
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/34da58ae/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
index 370a3eb..a00198d 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
@@ -81,7 +81,7 @@ class TaskResultGetterSuite extends FunSuite with BeforeAndAfter with BeforeAndA
test("handling results larger than Akka frame size") {
val akkaFrameSize =
- sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
+ sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.maximum-frame-size").toInt
val result = sc.parallelize(Seq(1), 1).map(x => 1.to(akkaFrameSize).toArray).reduce((x, y) => x)
assert(result === 1.to(akkaFrameSize).toArray)
@@ -102,7 +102,7 @@ class TaskResultGetterSuite extends FunSuite with BeforeAndAfter with BeforeAndA
}
scheduler.taskResultGetter = new ResultDeletingTaskResultGetter(sc.env, scheduler)
val akkaFrameSize =
- sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
+ sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.maximum-frame-size").toInt
val result = sc.parallelize(Seq(1), 1).map(x => 1.to(akkaFrameSize).toArray).reduce((x, y) => x)
assert(result === 1.to(akkaFrameSize).toArray)
[09/50] git commit: Fix API changes; lines > 100 chars.
Posted by ma...@apache.org.
Fix API changes; lines > 100 chars.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/b5e93c12
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/b5e93c12
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/b5e93c12
Branch: refs/heads/scala-2.10
Commit: b5e93c1227f0af965f15e9455e5f4bd72680ebde
Parents: 7d06bdd
Author: Harvey Feng <ha...@databricks.com>
Authored: Sat Oct 5 16:57:08 2013 -0700
Committer: Harvey Feng <ha...@databricks.com>
Committed: Sat Oct 5 16:57:08 2013 -0700
----------------------------------------------------------------------
.../main/scala/org/apache/spark/SparkContext.scala | 17 ++++++++++-------
.../scala/org/apache/spark/rdd/HadoopRDD.scala | 16 ++++++++--------
2 files changed, 18 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/b5e93c12/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala
index ada1037..df32a4a 100644
--- a/core/src/main/scala/org/apache/spark/SparkContext.scala
+++ b/core/src/main/scala/org/apache/spark/SparkContext.scala
@@ -84,9 +84,11 @@ class SparkContext(
val sparkHome: String = null,
val jars: Seq[String] = Nil,
val environment: Map[String, String] = Map(),
- // This is used only by yarn for now, but should be relevant to other cluster types (mesos, etc) too.
- // This is typically generated from InputFormatInfo.computePreferredLocations .. host, set of data-local splits on host
- val preferredNodeLocationData: scala.collection.Map[String, scala.collection.Set[SplitInfo]] = scala.collection.immutable.Map())
+ // This is used only by yarn for now, but should be relevant to other cluster types (mesos, etc)
+ // too. This is typically generated from InputFormatInfo.computePreferredLocations .. host, set
+ // of data-local splits on host
+ val preferredNodeLocationData: scala.collection.Map[String, scala.collection.Set[SplitInfo]] =
+ scala.collection.immutable.Map())
extends Logging {
// Ensure logging is initialized before we spawn any threads
@@ -239,7 +241,8 @@ class SparkContext(
val env = SparkEnv.get
val conf = env.hadoop.newConfiguration()
// Explicitly check for S3 environment variables
- if (System.getenv("AWS_ACCESS_KEY_ID") != null && System.getenv("AWS_SECRET_ACCESS_KEY") != null) {
+ if (System.getenv("AWS_ACCESS_KEY_ID") != null &&
+ System.getenv("AWS_SECRET_ACCESS_KEY") != null) {
conf.set("fs.s3.awsAccessKeyId", System.getenv("AWS_ACCESS_KEY_ID"))
conf.set("fs.s3n.awsAccessKeyId", System.getenv("AWS_ACCESS_KEY_ID"))
conf.set("fs.s3.awsSecretAccessKey", System.getenv("AWS_SECRET_ACCESS_KEY"))
@@ -332,15 +335,15 @@ class SparkContext(
* etc).
*/
def hadoopRDD[K, V](
- jobConf: JobConf,
+ conf: JobConf,
inputFormatClass: Class[_ <: InputFormat[K, V]],
keyClass: Class[K],
valueClass: Class[V],
minSplits: Int = defaultMinSplits
): RDD[(K, V)] = {
// Add necessary security credentials to the JobConf before broadcasting it.
- SparkEnv.get.hadoop.addCredentials(jobConf)
- new HadoopRDD(this, jobConf, inputFormatClass, keyClass, valueClass, minSplits)
+ SparkEnv.get.hadoop.addCredentials(conf)
+ new HadoopRDD(this, conf, inputFormatClass, keyClass, valueClass, minSplits)
}
/** Get an RDD for a Hadoop file with an arbitrary InputFormat */
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/b5e93c12/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
index 404532d..728f3d1 100644
--- a/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/HadoopRDD.scala
@@ -43,18 +43,18 @@ import org.apache.hadoop.conf.{Configuration, Configurable}
class HadoopFileRDD[K, V](
sc: SparkContext,
path: String,
- confBroadcast: Broadcast[SerializableWritable[Configuration]],
+ broadcastedConf: Broadcast[SerializableWritable[Configuration]],
inputFormatClass: Class[_ <: InputFormat[K, V]],
keyClass: Class[K],
valueClass: Class[V],
minSplits: Int)
- extends HadoopRDD[K, V](sc, confBroadcast, inputFormatClass, keyClass, valueClass, minSplits) {
+ extends HadoopRDD[K, V](sc, broadcastedConf, inputFormatClass, keyClass, valueClass, minSplits) {
override def getJobConf(): JobConf = {
if (HadoopRDD.containsCachedMetadata(jobConfCacheKey)) {
return HadoopRDD.getCachedMetadata(jobConfCacheKey).asInstanceOf[JobConf]
} else {
- val newJobConf = new JobConf(confBroadcast.value.value)
+ val newJobConf = new JobConf(broadcastedConf.value.value)
FileInputFormat.setInputPaths(newJobConf, path)
HadoopRDD.putCachedMetadata(jobConfCacheKey, newJobConf)
return newJobConf
@@ -80,7 +80,7 @@ private[spark] class HadoopPartition(rddId: Int, idx: Int, @transient s: InputSp
*/
class HadoopRDD[K, V](
sc: SparkContext,
- confBroadcast: Broadcast[SerializableWritable[Configuration]],
+ broadcastedConf: Broadcast[SerializableWritable[Configuration]],
inputFormatClass: Class[_ <: InputFormat[K, V]],
keyClass: Class[K],
valueClass: Class[V],
@@ -89,14 +89,14 @@ class HadoopRDD[K, V](
def this(
sc: SparkContext,
- jobConf: JobConf,
+ conf: JobConf,
inputFormatClass: Class[_ <: InputFormat[K, V]],
keyClass: Class[K],
valueClass: Class[V],
minSplits: Int) = {
this(
sc,
- sc.broadcast(new SerializableWritable(jobConf))
+ sc.broadcast(new SerializableWritable(conf))
.asInstanceOf[Broadcast[SerializableWritable[Configuration]]],
inputFormatClass,
keyClass,
@@ -110,13 +110,13 @@ class HadoopRDD[K, V](
// Returns a JobConf that will be used on slaves to obtain input splits for Hadoop reads.
protected def getJobConf(): JobConf = {
- val conf: Configuration = confBroadcast.value.value
+ val conf: Configuration = broadcastedConf.value.value
if (conf.isInstanceOf[JobConf]) {
return conf.asInstanceOf[JobConf]
} else if (HadoopRDD.containsCachedMetadata(jobConfCacheKey)) {
return HadoopRDD.getCachedMetadata(jobConfCacheKey).asInstanceOf[JobConf]
} else {
- val newJobConf = new JobConf(confBroadcast.value.value)
+ val newJobConf = new JobConf(broadcastedConf.value.value)
HadoopRDD.putCachedMetadata(jobConfCacheKey, newJobConf)
return newJobConf
}
[25/50] git commit: Responses to review
Posted by ma...@apache.org.
Responses to review
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/8b377718
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/8b377718
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/8b377718
Branch: refs/heads/scala-2.10
Commit: 8b377718b85d31fe5b0efb0ad77a8f38ffcede89
Parents: 391133f
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Oct 7 20:03:35 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Oct 7 20:03:35 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/CacheManager.scala | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/8b377718/core/src/main/scala/org/apache/spark/CacheManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/CacheManager.scala b/core/src/main/scala/org/apache/spark/CacheManager.scala
index 048168c..4cf7eb9 100644
--- a/core/src/main/scala/org/apache/spark/CacheManager.scala
+++ b/core/src/main/scala/org/apache/spark/CacheManager.scala
@@ -34,7 +34,7 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
def getOrCompute[T](rdd: RDD[T], split: Partition, context: TaskContext, storageLevel: StorageLevel)
: Iterator[T] = {
val key = "rdd_%d_%d".format(rdd.id, split.index)
- logInfo("Looking for partition " + key)
+ logDebug("Looking for partition " + key)
blockManager.get(key) match {
case Some(values) =>
// Partition is already materialized, so just return its values
@@ -44,11 +44,11 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
// Mark the split as loading (unless someone else marks it first)
loading.synchronized {
if (loading.contains(key)) {
- logInfo("Loading contains " + key + ", waiting...")
+ logInfo("Another thread is loading %s, waiting for it to finish...".format(key))
while (loading.contains(key)) {
try {loading.wait()} catch {case _ : Throwable =>}
}
- logInfo("Loading no longer contains " + key + ", so returning cached result")
+ logInfo("Finished waiting for %s".format(key))
// See whether someone else has successfully loaded it. The main way this would fail
// is for the RDD-level cache eviction policy if someone else has loaded the same RDD
// partition but we didn't want to make space for it. However, that case is unlikely
@@ -58,7 +58,7 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
case Some(values) =>
return values.asInstanceOf[Iterator[T]]
case None =>
- logInfo("Whoever was loading " + key + " failed; we'll try it ourselves")
+ logInfo("Whoever was loading %s failed; we'll try it ourselves".format(key))
loading.add(key)
}
} else {
[06/50] git commit: Fixed tests,
changed property akka.remote.netty.x to akka.remote.netty.tcp.x
Posted by ma...@apache.org.
Fixed tests, changed property akka.remote.netty.x to akka.remote.netty.tcp.x
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/3e414952
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/3e414952
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/3e414952
Branch: refs/heads/scala-2.10
Commit: 3e41495288c897ee3d3b31d5f4836b1cf6ba54d9
Parents: c810ee0
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Sat Oct 5 16:39:25 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Sat Oct 5 16:39:25 2013 +0530
----------------------------------------------------------------------
.../apache/spark/scheduler/cluster/TaskResultGetterSuite.scala | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/3e414952/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
index 119ba30..370a3eb 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/cluster/TaskResultGetterSuite.scala
@@ -81,7 +81,7 @@ class TaskResultGetterSuite extends FunSuite with BeforeAndAfter with BeforeAndA
test("handling results larger than Akka frame size") {
val akkaFrameSize =
- sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.message-frame-size").toInt
+ sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
val result = sc.parallelize(Seq(1), 1).map(x => 1.to(akkaFrameSize).toArray).reduce((x, y) => x)
assert(result === 1.to(akkaFrameSize).toArray)
@@ -102,7 +102,7 @@ class TaskResultGetterSuite extends FunSuite with BeforeAndAfter with BeforeAndA
}
scheduler.taskResultGetter = new ResultDeletingTaskResultGetter(sc.env, scheduler)
val akkaFrameSize =
- sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.message-frame-size").toInt
+ sc.env.actorSystem.settings.config.getBytes("akka.remote.netty.tcp.message-frame-size").toInt
val result = sc.parallelize(Seq(1), 1).map(x => 1.to(akkaFrameSize).toArray).reduce((x, y) => x)
assert(result === 1.to(akkaFrameSize).toArray)
[21/50] git commit: Adding Shark 0.7.1 to EC2 scripts
Posted by ma...@apache.org.
Adding Shark 0.7.1 to EC2 scripts
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/3745a182
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/3745a182
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/3745a182
Branch: refs/heads/scala-2.10
Commit: 3745a1827fc955be6c3236e4c31d27db062f15de
Parents: 213b70a
Author: Patrick Wendell <pw...@gmail.com>
Authored: Sat Oct 5 18:10:41 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Oct 7 15:03:42 2013 -0700
----------------------------------------------------------------------
ec2/spark_ec2.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/3745a182/ec2/spark_ec2.py
----------------------------------------------------------------------
diff --git a/ec2/spark_ec2.py b/ec2/spark_ec2.py
index 7da9355..65868b7 100755
--- a/ec2/spark_ec2.py
+++ b/ec2/spark_ec2.py
@@ -155,7 +155,7 @@ def is_active(instance):
# Return correct versions of Spark and Shark, given the supplied Spark version
def get_spark_shark_version(opts):
- spark_shark_map = {"0.7.3": "0.7.0", "0.8.0": "0.8.0"}
+ spark_shark_map = {"0.7.3": "0.7.1", "0.8.0": "0.8.0"}
version = opts.spark_version.replace("v", "")
if version not in spark_shark_map:
print >> stderr, "Don't know about Spark version: %s" % version
[49/50] git commit: Fixed some scala warnings in core.
Posted by ma...@apache.org.
Fixed some scala warnings in core.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/bfbd7e5d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/bfbd7e5d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/bfbd7e5d
Branch: refs/heads/scala-2.10
Commit: bfbd7e5d9f9e556961e79135dcde8f88fa31e2bc
Parents: 34da58a
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Thu Oct 10 15:22:31 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Thu Oct 10 15:22:31 2013 +0530
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala | 2 +-
.../org/apache/spark/scheduler/cluster/TaskResultGetter.scala | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/bfbd7e5d/core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala b/core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
index d7b45d4..c9c13f7 100644
--- a/core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
+++ b/core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
@@ -72,7 +72,7 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
* Return a new RDD by applying a function to each partition of this RDD, while tracking the index
* of the original partition.
*/
- def mapPartitionsWithIndex[R: ClassManifest](
+ def mapPartitionsWithIndex[R: ClassTag](
f: JFunction2[Int, java.util.Iterator[T], java.util.Iterator[R]],
preservesPartitioning: Boolean = false): JavaRDD[R] =
new JavaRDD(rdd.mapPartitionsWithIndex(((a,b) => f(a,asJavaIterator(b))),
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/bfbd7e5d/core/src/main/scala/org/apache/spark/scheduler/cluster/TaskResultGetter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/TaskResultGetter.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/TaskResultGetter.scala
index feec8ec..b2a8f06 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/TaskResultGetter.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/TaskResultGetter.scala
@@ -87,7 +87,7 @@ private[spark] class TaskResultGetter(sparkEnv: SparkEnv, scheduler: ClusterSche
case cnf: ClassNotFoundException =>
val loader = Thread.currentThread.getContextClassLoader
taskSetManager.abort("ClassNotFound with classloader: " + loader)
- case ex =>
+ case ex: Throwable =>
taskSetManager.abort("Exception while deserializing and fetching task: %s".format(ex))
}
}
@@ -111,7 +111,7 @@ private[spark] class TaskResultGetter(sparkEnv: SparkEnv, scheduler: ClusterSche
val loader = Thread.currentThread.getContextClassLoader
logError(
"Could not deserialize TaskEndReason: ClassNotFound with classloader " + loader)
- case ex => {}
+ case ex: Throwable => {}
}
scheduler.handleFailedTask(taskSetManager, tid, taskState, reason)
}
[31/50] git commit: remove those futile suffixes like number/count
Posted by ma...@apache.org.
remove those futile suffixes like number/count
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/f7628e40
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/f7628e40
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/f7628e40
Branch: refs/heads/scala-2.10
Commit: f7628e40330a1721e3d69ccb2390dfa75f839768
Parents: 22bed59
Author: Grace Huang <ji...@intel.com>
Authored: Wed Oct 9 08:36:41 2013 +0800
Committer: Grace Huang <ji...@intel.com>
Committed: Wed Oct 9 08:36:41 2013 +0800
----------------------------------------------------------------------
.../apache/spark/deploy/master/ApplicationSource.scala | 2 +-
.../org/apache/spark/deploy/master/MasterSource.scala | 6 +++---
.../org/apache/spark/deploy/worker/WorkerSource.scala | 6 +++---
.../scala/org/apache/spark/executor/ExecutorSource.scala | 4 ++--
.../org/apache/spark/scheduler/DAGSchedulerSource.scala | 10 +++++-----
5 files changed, 14 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
index 01dbb3d..c87b66f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/ApplicationSource.scala
@@ -34,7 +34,7 @@ class ApplicationSource(val application: ApplicationInfo) extends Source {
override def getValue: Long = application.duration
})
- metricRegistry.register(MetricRegistry.name("cores_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("cores"), new Gauge[Int] {
override def getValue: Int = application.coresGranted
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
index a0f6fc4..36c1b87 100644
--- a/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/master/MasterSource.scala
@@ -26,17 +26,17 @@ private[spark] class MasterSource(val master: Master) extends Source {
val sourceName = "master"
// Gauge for worker numbers in cluster
- metricRegistry.register(MetricRegistry.name("workers_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("workers"), new Gauge[Int] {
override def getValue: Int = master.workers.size
})
// Gauge for application numbers in cluster
- metricRegistry.register(MetricRegistry.name("apps_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("apps"), new Gauge[Int] {
override def getValue: Int = master.apps.size
})
// Gauge for waiting application numbers in cluster
- metricRegistry.register(MetricRegistry.name("waitingApps_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("waitingApps"), new Gauge[Int] {
override def getValue: Int = master.waitingApps.size
})
}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
index 5c2d3ee..b24e936 100644
--- a/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/worker/WorkerSource.scala
@@ -25,12 +25,12 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
val sourceName = "worker"
val metricRegistry = new MetricRegistry()
- metricRegistry.register(MetricRegistry.name("executors_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("executors"), new Gauge[Int] {
override def getValue: Int = worker.executors.size
})
// Gauge for cores used of this worker
- metricRegistry.register(MetricRegistry.name("coresUsed_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresUsed"), new Gauge[Int] {
override def getValue: Int = worker.coresUsed
})
@@ -40,7 +40,7 @@ private[spark] class WorkerSource(val worker: Worker) extends Source {
})
// Gauge for cores free of this worker
- metricRegistry.register(MetricRegistry.name("coresFree_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("coresFree"), new Gauge[Int] {
override def getValue: Int = worker.coresFree
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
index 2123904..34ed9c8 100644
--- a/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
+++ b/core/src/main/scala/org/apache/spark/executor/ExecutorSource.scala
@@ -43,12 +43,12 @@ class ExecutorSource(val executor: Executor, executorId: String) extends Source
val sourceName = "executor.%s".format(executorId)
// Gauge for executor thread pool's actively executing task counts
- metricRegistry.register(MetricRegistry.name("threadpool", "activeTask_count"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "activeTasks"), new Gauge[Int] {
override def getValue: Int = executor.threadPool.getActiveCount()
})
// Gauge for executor thread pool's approximate total number of tasks that have been completed
- metricRegistry.register(MetricRegistry.name("threadpool", "completeTask_count"), new Gauge[Long] {
+ metricRegistry.register(MetricRegistry.name("threadpool", "completeTasks"), new Gauge[Long] {
override def getValue: Long = executor.threadPool.getCompletedTaskCount()
})
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/f7628e40/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
index 474f3e8..1515148 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGSchedulerSource.scala
@@ -27,23 +27,23 @@ private[spark] class DAGSchedulerSource(val dagScheduler: DAGScheduler, sc: Spar
val metricRegistry = new MetricRegistry()
val sourceName = "%s.DAGScheduler".format(sc.appName)
- metricRegistry.register(MetricRegistry.name("stage", "failedStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "failedStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.failed.size
})
- metricRegistry.register(MetricRegistry.name("stage", "runningStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "runningStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.running.size
})
- metricRegistry.register(MetricRegistry.name("stage", "waitingStages_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("stage", "waitingStages"), new Gauge[Int] {
override def getValue: Int = dagScheduler.waiting.size
})
- metricRegistry.register(MetricRegistry.name("job", "allJobs_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "allJobs"), new Gauge[Int] {
override def getValue: Int = dagScheduler.nextJobId.get()
})
- metricRegistry.register(MetricRegistry.name("job", "activeJobs_number"), new Gauge[Int] {
+ metricRegistry.register(MetricRegistry.name("job", "activeJobs"), new Gauge[Int] {
override def getValue: Int = dagScheduler.activeJobs.size
})
}
[47/50] git commit: Merge branch 'master' of
github.com:apache/incubator-spark into scala-2.10
Posted by ma...@apache.org.
Merge branch 'master' of github.com:apache/incubator-spark into scala-2.10
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/026ab756
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/026ab756
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/026ab756
Branch: refs/heads/scala-2.10
Commit: 026ab7566167e6c8ab1b0cce75b9e09bbd485bee
Parents: 2686063 320418f
Author: Prashant Sharma <pr...@imaginea.com>
Authored: Thu Oct 10 09:42:55 2013 +0530
Committer: Prashant Sharma <pr...@imaginea.com>
Committed: Thu Oct 10 09:42:55 2013 +0530
----------------------------------------------------------------------
.../scala/org/apache/spark/SparkContext.scala | 2 +-
.../spark/deploy/master/ApplicationSource.scala | 2 +-
.../spark/deploy/master/MasterSource.scala | 6 +-
.../spark/deploy/worker/WorkerSource.scala | 10 +-
.../apache/spark/executor/ExecutorSource.scala | 18 +-
.../apache/spark/scheduler/DAGScheduler.scala | 2 +-
.../spark/scheduler/DAGSchedulerSource.scala | 10 +-
.../spark/scheduler/SparkListenerBus.scala | 18 ++
.../spark/serializer/KryoSerializer.scala | 22 +-
.../spark/storage/BlockManagerSource.scala | 8 +-
.../spark/scheduler/SparkListenerSuite.scala | 32 +--
.../spark/serializer/KryoSerializerSuite.scala | 21 ++
docs/mllib-guide.md | 24 ++-
docs/python-programming-guide.md | 2 +-
.../apache/spark/mllib/recommendation/ALS.scala | 199 ++++++++++++++++---
.../mllib/recommendation/JavaALSSuite.java | 85 ++++++--
.../spark/mllib/recommendation/ALSSuite.scala | 75 +++++--
python/pyspark/rdd.py | 60 +++++-
18 files changed, 474 insertions(+), 122 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/026ab756/core/src/main/scala/org/apache/spark/SparkContext.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/026ab756/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/026ab756/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
----------------------------------------------------------------------
diff --cc mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
index c474e01,eafee06..b40f552
--- a/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
+++ b/mllib/src/test/java/org/apache/spark/mllib/recommendation/JavaALSSuite.java
@@@ -19,7 -19,10 +19,8 @@@ package org.apache.spark.mllib.recommen
import java.io.Serializable;
import java.util.List;
+ import java.lang.Math;
-import scala.Tuple2;
-
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/026ab756/python/pyspark/rdd.py
----------------------------------------------------------------------
[45/50] git commit: Merge pull request #49 from mateiz/kryo-fix-2
Posted by ma...@apache.org.
Merge pull request #49 from mateiz/kryo-fix-2
Fix Chill serialization of Range objects
It used to write out each element one by one, creating very large objects.
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/320418f7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/320418f7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/320418f7
Branch: refs/heads/scala-2.10
Commit: 320418f7c8b42d4ce781b32c9ee47a9b54550b89
Parents: 215238c c84c205
Author: Reynold Xin <rx...@apache.org>
Authored: Wed Oct 9 16:55:30 2013 -0700
Committer: Reynold Xin <rx...@apache.org>
Committed: Wed Oct 9 16:55:30 2013 -0700
----------------------------------------------------------------------
.../spark/serializer/KryoSerializer.scala | 14 ++++++++++---
.../spark/serializer/KryoSerializerSuite.scala | 21 ++++++++++++++++++++
2 files changed, 32 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
[23/50] git commit: Fix inconsistent and incorrect log messages in
shuffle read path
Posted by ma...@apache.org.
Fix inconsistent and incorrect log messages in shuffle read path
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/391133f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/391133f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/391133f6
Branch: refs/heads/scala-2.10
Commit: 391133f66a41cf78cc200c20c0228eb99eebc6fd
Parents: 3745a18
Author: Patrick Wendell <pw...@gmail.com>
Authored: Mon Oct 7 17:08:06 2013 -0700
Committer: Patrick Wendell <pw...@gmail.com>
Committed: Mon Oct 7 17:24:18 2013 -0700
----------------------------------------------------------------------
core/src/main/scala/org/apache/spark/CacheManager.scala | 11 +++++------
.../scala/org/apache/spark/storage/BlockManager.scala | 12 +++++++++++-
2 files changed, 16 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/391133f6/core/src/main/scala/org/apache/spark/CacheManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/CacheManager.scala b/core/src/main/scala/org/apache/spark/CacheManager.scala
index 3d36761..048168c 100644
--- a/core/src/main/scala/org/apache/spark/CacheManager.scala
+++ b/core/src/main/scala/org/apache/spark/CacheManager.scala
@@ -34,12 +34,11 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
def getOrCompute[T](rdd: RDD[T], split: Partition, context: TaskContext, storageLevel: StorageLevel)
: Iterator[T] = {
val key = "rdd_%d_%d".format(rdd.id, split.index)
- logInfo("Cache key is " + key)
+ logInfo("Looking for partition " + key)
blockManager.get(key) match {
- case Some(cachedValues) =>
- // Partition is in cache, so just return its values
- logInfo("Found partition in cache!")
- return cachedValues.asInstanceOf[Iterator[T]]
+ case Some(values) =>
+ // Partition is already materialized, so just return its values
+ return values.asInstanceOf[Iterator[T]]
case None =>
// Mark the split as loading (unless someone else marks it first)
@@ -68,7 +67,7 @@ private[spark] class CacheManager(blockManager: BlockManager) extends Logging {
}
try {
// If we got here, we have to load the split
- logInfo("Computing partition " + split)
+ logInfo("Partition %s not found, computing it".format(key))
val computedValues = rdd.computeOrReadCheckpoint(split, context)
// Persist the result, so long as the task is not running locally
if (context.runningLocally) { return computedValues }
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/391133f6/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
index 495a72d..37d0ddb 100644
--- a/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
+++ b/core/src/main/scala/org/apache/spark/storage/BlockManager.scala
@@ -523,7 +523,17 @@ private[spark] class BlockManager(
* Get a block from the block manager (either local or remote).
*/
def get(blockId: String): Option[Iterator[Any]] = {
- getLocal(blockId).orElse(getRemote(blockId))
+ val local = getLocal(blockId)
+ if (local.isDefined) {
+ logInfo("Found block %s locally".format(blockId))
+ return local
+ }
+ val remote = getRemote(blockId)
+ if (remote.isDefined) {
+ logInfo("Found block %s remotely".format(blockId))
+ return remote
+ }
+ None
}
/**