You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mahout.apache.org by dl...@apache.org on 2014/04/22 20:39:31 UTC
svn commit: r1589246 - in /mahout/trunk: ./ bin/ math-scala/ spark-shell/
spark-shell/src/ spark-shell/src/main/ spark-shell/src/main/scala/
spark-shell/src/main/scala/org/ spark-shell/src/main/scala/org/apache/
spark-shell/src/main/scala/org/apache/ma...
Author: dlyubimov
Date: Tue Apr 22 18:39:30 2014
New Revision: 1589246
URL: http://svn.apache.org/r1589246
Log:
MAHOUT-1489 : initial Mahout spark shell commit
Squashed commit of the following:
commit 0124072b72fcdad9ccded43745c9b1d00e7ea089
Merge: c1a2c8a c9164c1
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Tue Apr 22 11:33:17 2014 -0700
Merge branch 'trunk' into shell
commit c1a2c8a414c015dcdce592b145498fc8b836addf
Merge: a3491f5 a8df05b
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Tue Apr 22 11:30:25 2014 -0700
Merge branch 'trunk' into shell; misc fixes
Conflicts:
bin/mahout
commit a3491f57e77b4b789051cf131bc7fdea73ad3e41
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Mon Apr 21 17:39:06 2014 -0700
-NonLocal
commit bd0c83ebfa66e48f434f0ecc81bf81dd07d27f8c
Merge: ad01add 78c45c4
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Mon Apr 21 17:33:18 2014 -0700
Merge commit '78c45c4c5d96f51e9' into shell
commit ad01add55c1c5a212dabcd727e31dd36162c1fd0
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Mon Apr 21 17:30:53 2014 -0700
Fixing writeDRM problems
commit f9f20e364e4462b6ba0693d11ee939ef5afd43a4
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Mon Apr 21 14:58:28 2014 -0700
writeDRM broken, some unknown closure attributes yank "this", not seeing where and which
commit 7ca93f70ef7313cca251470c41032c08e9e612e2
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Mon Apr 21 12:57:40 2014 -0700
script errors
commit ce5bdfaba160ecb6ee433b4682f62d9e6236e0b6
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Thu Apr 17 13:51:35 2014 -0700
drmFromHDFS() fixes -- load and initiailize classtag evidence correctly.
TODO: use class evidence from key for saveDRM.
commit aeba609a957b9ef5aac7aa4eba01edb46004116f
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Thu Apr 17 11:41:30 2014 -0700
renaming script extensions
commit 9b02ac2b6aa4ff046a898876f9a8fb23936cd78b
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Thu Apr 17 11:17:04 2014 -0700
build fix
commit 2fbd835c4c4f8fac3ff32d06b9578d436f4bbf09
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Wed Apr 16 18:04:47 2014 -0700
WIP -- unstable
commit cc87347d393709ac0a6ab2adc66200caed07e911
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Wed Apr 16 15:27:00 2014 -0700
removing examples dependencies if running spark shell.
commit 435992aa99090b9a75f3903ac5d10e43e6b49357
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Wed Apr 16 14:10:22 2014 -0700
WIP spark-shell, seems to be working
commit 1f4fd51c2e5e18852d1b30d5a88897e14761d9e8
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Wed Apr 16 12:55:17 2014 -0700
WIP -- implementing "mathout spark-shell"
commit 5922a111405cc66c1d598b123c7d28f456229559
Author: Dmitriy Lyubimov <dl...@apache.org>
Date: Thu Apr 10 17:53:23 2014 -0700
First shell prototype works in local mode
Added:
mahout/trunk/spark-shell/
mahout/trunk/spark-shell/pom.xml
- copied, changed from r1589027, mahout/trunk/math-scala/pom.xml
mahout/trunk/spark-shell/src/
mahout/trunk/spark-shell/src/main/
mahout/trunk/spark-shell/src/main/scala/
mahout/trunk/spark-shell/src/main/scala/org/
mahout/trunk/spark-shell/src/main/scala/org/apache/
mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/
mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/
mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/
mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/MahoutSparkILoop.scala
mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala
mahout/trunk/spark-shell/src/test/
mahout/trunk/spark-shell/src/test/mahout/
mahout/trunk/spark-shell/src/test/mahout/simple.mscala
Modified:
mahout/trunk/bin/mahout
mahout/trunk/math-scala/pom.xml
mahout/trunk/pom.xml
mahout/trunk/spark/pom.xml
mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrm.scala
mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrmBase.scala
mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/package.scala
mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/package.scala
mahout/trunk/spark/src/test/scala/org/apache/mahout/sparkbindings/test/MahoutLocalContext.scala
Modified: mahout/trunk/bin/mahout
URL: http://svn.apache.org/viewvc/mahout/trunk/bin/mahout?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/bin/mahout (original)
+++ mahout/trunk/bin/mahout Tue Apr 22 18:39:30 2014
@@ -75,6 +75,16 @@ if [ "$1" == "-core" ] ; then
shift
fi
+if [ "$1" == "-spark" ]; then
+ SPARK=1
+ shift
+fi
+
+if [ "$1" == "spark-shell" ]; then
+ SPARK=1
+fi
+
+
if [ "$MAHOUT_CORE" != "" ]; then
IS_CORE=1
fi
@@ -116,6 +126,7 @@ if [ "x$MAHOUT_CONF_DIR" = "x" ]; then
fi
fi
+
# CLASSPATH initially contains $MAHOUT_CONF_DIR, or defaults to $MAHOUT_HOME/src/conf
CLASSPATH=${CLASSPATH}:$MAHOUT_CONF_DIR
@@ -138,21 +149,45 @@ then
CLASSPATH=${CLASSPATH}:$f;
done
- # add dev targets if they exist
- for f in $MAHOUT_HOME/examples/target/mahout-examples-*-job.jar $MAHOUT_HOME/mahout-examples-*-job.jar ; do
- CLASSPATH=${CLASSPATH}:$f;
- done
+ if [ "$SPARK" != "1" ]; then
+
+ # add dev targets if they exist
+ for f in $MAHOUT_HOME/examples/target/mahout-examples-*-job.jar $MAHOUT_HOME/mahout-examples-*-job.jar ; do
+ CLASSPATH=${CLASSPATH}:$f;
+ done
+ fi
- # add scala dev target since it is not currently used in examples
+ # add scala dev target
for f in $MAHOUT_HOME/math-scala/target/mahout-math-scala-*.jar ; do
CLASSPATH=${CLASSPATH}:$f;
done
- # add mahout-spark since it is currently not used in examples
- for f in $MAHOUT_HOME/spark/target/mahout-spark-*.jar ; do
- CLASSPATH=${CLASSPATH}:$f;
- done
-
+
+ # add spark-shell -- if we requested shell
+ if [ "$SPARK" == "1" ]; then
+
+ for f in $MAHOUT_HOME/core/target/mahout-core-*.jar ; do
+ CLASSPATH=${CLASSPATH}:$f;
+ done
+
+ for f in $MAHOUT_HOME/spark/target/mahout-spark-*.jar ; do
+ CLASSPATH=${CLASSPATH}:$f;
+ done
+
+ for f in $MAHOUT_HOME/spark-shell/target/mahout-spark-shell-*.jar ; do
+ CLASSPATH=${CLASSPATH}:$f;
+ done
+
+ SPARK_CP_BIN="${SPARK_HOME}/bin/compute-classpath.sh"
+ if [ -x "${SPARK_CP_BIN}" ]; then
+ SPARK_CLASSPATH=$("${SPARK_CP_BIN}" 2>/dev/null)
+ CLASSPATH="${CLASSPATH}:${SPARK_CLASSPATH}"
+ else
+ echo "Cannot find Spark classpath."
+ exit -1
+ fi
+
+ fi
# add release dependencies to CLASSPATH
for f in $MAHOUT_HOME/lib/*.jar; do
@@ -165,12 +200,16 @@ else
CLASSPATH=${CLASSPATH}:$MAHOUT_HOME/examples/target/classes
CLASSPATH=${CLASSPATH}:$MAHOUT_HOME/math-scala/target/classes
#CLASSPATH=${CLASSPATH}:$MAHOUT_HOME/mrlegacy/src/main/resources
+ CLASSPATH=${CLASSPATH}:$MAHOUT_HOME/spark/target/classes
+ CLASSPATH=${CLASSPATH}:$MAHOUT_HOME/spark-shell/target/classes
fi
# add development dependencies to CLASSPATH
-for f in $MAHOUT_HOME/examples/target/dependency/*.jar; do
- CLASSPATH=${CLASSPATH}:$f;
-done
+if [ "$SPARK" == "" ]; then
+ for f in $MAHOUT_HOME/examples/target/dependency/*.jar; do
+ CLASSPATH=${CLASSPATH}:$f;
+ done
+fi
# cygwin path translation
@@ -181,85 +220,95 @@ fi
# restore ordinary behaviour
unset IFS
-# default log directory & file
-if [ "$MAHOUT_LOG_DIR" = "" ]; then
- MAHOUT_LOG_DIR="$MAHOUT_HOME/logs"
-fi
-if [ "$MAHOUT_LOGFILE" = "" ]; then
- MAHOUT_LOGFILE='mahout.log'
-fi
-#Fix log path under cygwin
-if $cygwin; then
- MAHOUT_LOG_DIR=`cygpath -p -w "$MAHOUT_LOG_DIR"`
-fi
-MAHOUT_OPTS="$MAHOUT_OPTS -Dhadoop.log.dir=$MAHOUT_LOG_DIR"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dhadoop.log.file=$MAHOUT_LOGFILE"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.min.split.size=512MB"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.map.child.java.opts=-Xmx4096m"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.reduce.child.java.opts=-Xmx4096m"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.output.compress=true"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.compress.map.output=true"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.map.tasks=1"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.reduce.tasks=1"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dio.sort.factor=30"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dio.sort.mb=1024"
-MAHOUT_OPTS="$MAHOUT_OPTS -Dio.file.buffer.size=32786"
-
-if [ "x$JAVA_LIBRARY_PATH" != "x" ]; then
- MAHOUT_OPTS="$MAHOUT_OPTS -Djava.library.path=$JAVA_LIBRARY_PATH"
-fi
-
-CLASS=org.apache.mahout.driver.MahoutDriver
-
-for f in $MAHOUT_HOME/examples/target/mahout-examples-*-job.jar $MAHOUT_HOME/mahout-examples-*-job.jar ; do
- if [ -e "$f" ]; then
- MAHOUT_JOB=$f
- fi
-done
+case "$1" in
+ (spark-shell)
+ exec "$JAVA" $JAVA_HEAP_MAX -classpath "$CLASSPATH" "org.apache.mahout.sparkbindings.shell.Main" $@
+ ;;
+ (*)
+
+ # default log directory & file
+ if [ "$MAHOUT_LOG_DIR" = "" ]; then
+ MAHOUT_LOG_DIR="$MAHOUT_HOME/logs"
+ fi
+ if [ "$MAHOUT_LOGFILE" = "" ]; then
+ MAHOUT_LOGFILE='mahout.log'
+ fi
-# run it
+ #Fix log path under cygwin
+ if $cygwin; then
+ MAHOUT_LOG_DIR=`cygpath -p -w "$MAHOUT_LOG_DIR"`
+ fi
-HADOOP_BINARY=$(PATH="${HADOOP_HOME:-${HADOOP_PREFIX}}/bin:$PATH" which hadoop 2>/dev/null)
-if [ -x "$HADOOP_BINARY" ] ; then
- HADOOP_BINARY_CLASSPATH=$("$HADOOP_BINARY" classpath)
-fi
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dhadoop.log.dir=$MAHOUT_LOG_DIR"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dhadoop.log.file=$MAHOUT_LOGFILE"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.min.split.size=512MB"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.map.child.java.opts=-Xmx4096m"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.reduce.child.java.opts=-Xmx4096m"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.output.compress=true"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.compress.map.output=true"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.map.tasks=1"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dmapred.reduce.tasks=1"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dio.sort.factor=30"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dio.sort.mb=1024"
+ MAHOUT_OPTS="$MAHOUT_OPTS -Dio.file.buffer.size=32786"
-if [ ! -x "$HADOOP_BINARY" ] || [ "$MAHOUT_LOCAL" != "" ] ; then
- if [ ! -x "$HADOOP_BINARY" ] ; then
- echo "hadoop binary is not in PATH,HADOOP_HOME/bin,HADOOP_PREFIX/bin, running locally"
- elif [ "$MAHOUT_LOCAL" != "" ] ; then
- echo "MAHOUT_LOCAL is set, running locally"
- fi
- CLASSPATH="${CLASSPATH}:${MAHOUT_HOME}/lib/hadoop/*"
- case $1 in
- (classpath)
- echo $CLASSPATH
- ;;
- (*)
- exec "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH" $CLASS "$@"
- esac
-else
- echo "Running on hadoop, using $HADOOP_BINARY and HADOOP_CONF_DIR=$HADOOP_CONF_DIR"
+ if [ "x$JAVA_LIBRARY_PATH" != "x" ]; then
+ MAHOUT_OPTS="$MAHOUT_OPTS -Djava.library.path=$JAVA_LIBRARY_PATH"
+ fi
+
+ CLASS=org.apache.mahout.driver.MahoutDriver
+
+ for f in $MAHOUT_HOME/examples/target/mahout-examples-*-job.jar $MAHOUT_HOME/mahout-examples-*-job.jar ; do
+ if [ -e "$f" ]; then
+ MAHOUT_JOB=$f
+ fi
+ done
+
+ # run it
+
+ HADOOP_BINARY=$(PATH="${HADOOP_HOME:-${HADOOP_PREFIX}}/bin:$PATH" which hadoop 2>/dev/null)
+ if [ -x "$HADOOP_BINARY" ] ; then
+ HADOOP_BINARY_CLASSPATH=$("$HADOOP_BINARY" classpath)
+ fi
+ if [ ! -x "$HADOOP_BINARY" ] || [ "$MAHOUT_LOCAL" != "" ] ; then
+ if [ ! -x "$HADOOP_BINARY" ] ; then
+ echo "hadoop binary is not in PATH,HADOOP_HOME/bin,HADOOP_PREFIX/bin, running locally"
+ elif [ "$MAHOUT_LOCAL" != "" ] ; then
+ echo "MAHOUT_LOCAL is set, running locally"
+ fi
+ CLASSPATH="${CLASSPATH}:${MAHOUT_HOME}/lib/hadoop/*"
+ case $1 in
+ (classpath)
+ echo $CLASSPATH
+ ;;
+ (*)
+ exec "$JAVA" $JAVA_HEAP_MAX $MAHOUT_OPTS -classpath "$CLASSPATH" $CLASS "$@"
+ esac
+ else
+ echo "Running on hadoop, using $HADOOP_BINARY and HADOOP_CONF_DIR=$HADOOP_CONF_DIR"
+
+ if [ "$MAHOUT_JOB" = "" ] ; then
+ echo "ERROR: Could not find mahout-examples-*.job in $MAHOUT_HOME or $MAHOUT_HOME/examples/target, please run 'mvn install' to create the .job file"
+ exit 1
+ else
+ case "$1" in
+ (hadoop)
+ shift
+ export HADOOP_CLASSPATH=$MAHOUT_CONF_DIR:${HADOOP_CLASSPATH}:$CLASSPATH
+ exec "$HADOOP_BINARY" "$@"
+ ;;
+ (classpath)
+ echo $CLASSPATH
+ ;;
+ (*)
+ echo "MAHOUT-JOB: $MAHOUT_JOB"
+ export HADOOP_CLASSPATH=$MAHOUT_CONF_DIR:${HADOOP_CLASSPATH}
+ exec "$HADOOP_BINARY" jar $MAHOUT_JOB $CLASS "$@"
+ esac
+ fi
+ fi
+ ;;
+esac
- if [ "$MAHOUT_JOB" = "" ] ; then
- echo "ERROR: Could not find mahout-examples-*.job in $MAHOUT_HOME or $MAHOUT_HOME/examples/target, please run 'mvn install' to create the .job file"
- exit 1
- else
- case "$1" in
- (hadoop)
- shift
- export HADOOP_CLASSPATH=$MAHOUT_CONF_DIR:${HADOOP_CLASSPATH}:$CLASSPATH
- exec "$HADOOP_BINARY" "$@"
- ;;
- (classpath)
- echo $CLASSPATH
- ;;
- (*)
- echo "MAHOUT-JOB: $MAHOUT_JOB"
- export HADOOP_CLASSPATH=$MAHOUT_CONF_DIR:${HADOOP_CLASSPATH}
- exec "$HADOOP_BINARY" jar $MAHOUT_JOB $CLASS "$@"
- esac
- fi
-fi
Modified: mahout/trunk/math-scala/pom.xml
URL: http://svn.apache.org/viewvc/mahout/trunk/math-scala/pom.xml?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/math-scala/pom.xml (original)
+++ mahout/trunk/math-scala/pom.xml Tue Apr 22 18:39:30 2014
@@ -178,12 +178,12 @@
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
- <version>2.10.3</version>
+ <version>${scala.version}</version>
</dependency>
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_2.10</artifactId>
+ <artifactId>scalatest_${scala.major}</artifactId>
<version>2.0</version>
<scope>test</scope>
</dependency>
Modified: mahout/trunk/pom.xml
URL: http://svn.apache.org/viewvc/mahout/trunk/pom.xml?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/pom.xml (original)
+++ mahout/trunk/pom.xml Tue Apr 22 18:39:30 2014
@@ -107,6 +107,9 @@
<hadoop.1.version>1.2.1</hadoop.1.version>
<lucene.version>4.6.1</lucene.version>
<slf4j.version>1.7.5</slf4j.version>
+ <scala.major>2.10</scala.major>
+ <scala.version>2.10.3</scala.version>
+ <spark.version>0.9.1</spark.version>
</properties>
<issueManagement>
<system>Jira</system>
@@ -673,6 +676,7 @@
<module>distribution</module>
<module>math-scala</module>
<module>spark</module>
+ <module>spark-shell</module>
</modules>
<profiles>
<profile>
Copied: mahout/trunk/spark-shell/pom.xml (from r1589027, mahout/trunk/math-scala/pom.xml)
URL: http://svn.apache.org/viewvc/mahout/trunk/spark-shell/pom.xml?p2=mahout/trunk/spark-shell/pom.xml&p1=mahout/trunk/math-scala/pom.xml&r1=1589027&r2=1589246&rev=1589246&view=diff
==============================================================================
--- mahout/trunk/math-scala/pom.xml (original)
+++ mahout/trunk/spark-shell/pom.xml Tue Apr 22 18:39:30 2014
@@ -17,7 +17,8 @@
limitations under the License.
-->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
@@ -27,11 +28,10 @@
<relativePath>../pom.xml</relativePath>
</parent>
- <artifactId>mahout-math-scala</artifactId>
- <name>Mahout Math/Scala wrappers</name>
- <description>High performance scientific and technical computing data structures and methods,
- mostly based on CERN's
- Colt Java API
+ <artifactId>mahout-spark-shell</artifactId>
+ <name>Mahout Spark bindings shell</name>
+ <description>
+ Mahout Bindings for Apache Spark
</description>
<packaging>jar</packaging>
@@ -108,7 +108,6 @@
<plugin>
<groupId>org.scala-tools</groupId>
<artifactId>maven-scala-plugin</artifactId>
- <version>2.15.2</version>
<executions>
<execution>
<goals>
@@ -130,12 +129,12 @@
<!-- disable surefire -->
<!--<plugin>-->
- <!--<groupId>org.apache.maven.plugins</groupId>-->
- <!--<artifactId>maven-surefire-plugin</artifactId>-->
- <!--<version>2.7</version>-->
- <!--<configuration>-->
- <!--<skipTests>true</skipTests>-->
- <!--</configuration>-->
+ <!--<groupId>org.apache.maven.plugins</groupId>-->
+ <!--<artifactId>maven-surefire-plugin</artifactId>-->
+ <!--<version>2.7</version>-->
+ <!--<configuration>-->
+ <!--<skipTests>true</skipTests>-->
+ <!--</configuration>-->
<!--</plugin>-->
<!-- enable scalatest -->
<plugin>
@@ -162,25 +161,33 @@
<dependencies>
+
+
<dependency>
<groupId>org.apache.mahout</groupId>
- <artifactId>mahout-math</artifactId>
+ <artifactId>mahout-spark</artifactId>
+ <version>${pom.version}</version>
</dependency>
-
- <!-- 3rd-party -->
<dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
+ <groupId>org.apache.mahout</groupId>
+ <artifactId>mahout-math-scala</artifactId>
+ <classifier>tests</classifier>
+ <version>${pom.version}</version>
+ <scope>test</scope>
</dependency>
- <!-- scala stuff -->
+
+ <!-- 3rd-party -->
+
+ <!-- Spark stuff -->
<dependency>
- <groupId>org.scala-lang</groupId>
- <artifactId>scala-library</artifactId>
- <version>2.10.3</version>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-repl_${scala.major}</artifactId>
+ <version>${spark.version}</version>
</dependency>
+ <!-- scala stuff -->
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_2.10</artifactId>
Added: mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/MahoutSparkILoop.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/MahoutSparkILoop.scala?rev=1589246&view=auto
==============================================================================
--- mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/MahoutSparkILoop.scala (added)
+++ mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/MahoutSparkILoop.scala Tue Apr 22 18:39:30 2014
@@ -0,0 +1,54 @@
+package org.apache.mahout.sparkbindings.shell
+
+import org.apache.spark.repl.SparkILoop
+import org.apache.spark.{SparkConf, SparkContext}
+import org.apache.mahout.sparkbindings._
+
+class MahoutSparkILoop extends SparkILoop {
+
+ private val postInitScript =
+ "import org.apache.mahout.math.scalabindings._" ::
+ "import RLikeOps._" ::
+ "import org.apache.mahout.sparkbindings._" ::
+ "import drm._" ::
+ "import RLikeDrmOps._" ::
+ "org.apache.spark.storage.StorageLevel" ::
+ "implicit val _sc = sc" ::
+ Nil
+
+ override protected def postInitialization() {
+ super.postInitialization()
+ val intp: MahoutSparkILoop = this
+ intp.beQuietDuring {
+ postInitScript.foreach(command(_))
+ }
+ }
+
+ override def createSparkContext(): SparkContext = {
+ val execUri = System.getenv("SPARK_EXECUTOR_URI")
+ val master = this.master match {
+ case Some(m) => m
+ case None => {
+ val prop = System.getenv("MASTER")
+ if (prop != null) prop else "local"
+ }
+ }
+ val jars = SparkILoop.getAddedJars.map(new java.io.File(_).getAbsolutePath)
+ val conf = new SparkConf()
+ .set("spark.repl.class.uri", intp.classServer.uri)
+
+ if (execUri != null) {
+ conf.set("spark.executor.uri", execUri)
+ }
+
+ sparkContext = mahoutSparkContext(
+ masterUrl = master,
+ appName = "Mahout Spark Shell",
+ customJars = jars,
+ sparkConf = conf
+ )
+ echo("Created spark context..")
+ sparkContext
+ }
+
+}
Added: mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala?rev=1589246&view=auto
==============================================================================
--- mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala (added)
+++ mahout/trunk/spark-shell/src/main/scala/org/apache/mahout/sparkbindings/shell/Main.scala Tue Apr 22 18:39:30 2014
@@ -0,0 +1,17 @@
+package org.apache.mahout.sparkbindings.shell
+
+
+object Main {
+
+ private var _interp: MahoutSparkILoop = _
+
+ def main(args:Array[String]) {
+ System.setProperty("scala.usejavacp", "true")
+ _interp = new MahoutSparkILoop()
+ // It looks like we need to initialize this too, since some Spark shell initilaization code
+ // expects it
+ org.apache.spark.repl.Main.interp = _interp
+ _interp.process(args)
+ }
+
+}
Added: mahout/trunk/spark-shell/src/test/mahout/simple.mscala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark-shell/src/test/mahout/simple.mscala?rev=1589246&view=auto
==============================================================================
--- mahout/trunk/spark-shell/src/test/mahout/simple.mscala (added)
+++ mahout/trunk/spark-shell/src/test/mahout/simple.mscala Tue Apr 22 18:39:30 2014
@@ -0,0 +1,23 @@
+import org.apache.mahout.sparkbindings._
+import drm._
+import RLikeDrmOps._
+
+import org.apache.spark.storage.StorageLevel
+
+val a = dense((1,2,3),(3,4,5))
+val drmA = drmParallelize(a,numPartitions = 2)
+val drmAtA = drmA.t %*% drmA
+
+val r = drmAtA.mapBlock() {
+ case (keys, block) =>
+ block += 1.0
+ keys -> block
+}.checkpoint(/*StorageLevel.NONE*/)
+
+r.collect
+
+// local write
+r.writeDRM("file:///home/dmitriy/A")
+
+// hdfs write
+r.writeDRM("hdfs://localhost:11010/A")
\ No newline at end of file
Modified: mahout/trunk/spark/pom.xml
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/pom.xml?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/pom.xml (original)
+++ mahout/trunk/spark/pom.xml Tue Apr 22 18:39:30 2014
@@ -213,23 +213,16 @@
</dependencies>
</profile>
</profiles>
- <dependencies>
+ <dependencies>
<!-- spark stuff -->
<dependency>
<groupId>org.apache.spark</groupId>
- <artifactId>spark-core_2.10</artifactId>
- <version>0.9.0-incubating</version>
+ <artifactId>spark-core_${scala.major}</artifactId>
+ <version>${spark.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-bagel_2.10</artifactId>
- <version>0.9.0-incubating</version>
- </dependency>
-
- <dependency>
-
<groupId>org.apache.mahout</groupId>
<artifactId>mahout-mrlegacy</artifactId>
<version>${pom.version}</version>
@@ -263,7 +256,7 @@
<dependency>
<groupId>org.scalatest</groupId>
- <artifactId>scalatest_2.10</artifactId>
+ <artifactId>scalatest_${scala.major}</artifactId>
<version>2.0</version>
<scope>test</scope>
</dependency>
Modified: mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrm.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrm.scala?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrm.scala (original)
+++ mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrm.scala Tue Apr 22 18:39:30 2014
@@ -31,7 +31,7 @@ trait CheckpointedDrm[K] extends DrmLike
def collect: Matrix
- def writeDRM(path: String)(implicit ev: K => Writable)
+ def writeDRM(path: String)
/** If this checkpoint is already declared cached, uncache. */
def uncache()
Modified: mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrmBase.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrmBase.scala?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrmBase.scala (original)
+++ mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/CheckpointedDrmBase.scala Tue Apr 22 18:39:30 2014
@@ -23,17 +23,16 @@ import org.apache.mahout.math.scalabindi
import RLikeOps._
import scala.collection.JavaConversions._
import org.apache.spark.storage.StorageLevel
-import org.apache.spark.rdd.RDD
-import org.apache.hadoop.io.Writable
import org.apache.spark.SparkContext._
import reflect._
import scala.util.Random
+import org.apache.hadoop.io.{LongWritable, Text, IntWritable, Writable}
/**
*
* @author dmitriy
*/
-class CheckpointedDrmBase[K : ClassTag](
+class CheckpointedDrmBase[K: ClassTag](
val rdd: DrmRdd[K],
private var _nrow: Long = -1L,
private var _ncol: Int = -1,
@@ -53,7 +52,7 @@ class CheckpointedDrmBase[K : ClassTag](
* Action operator -- does not necessary means Spark action; but does mean running BLAS optimizer
* and writing down Spark graph lineage since last checkpointed DRM.
*/
- def checkpoint(sLevel:StorageLevel): CheckpointedDrm[K] =
+ def checkpoint(sLevel: StorageLevel): CheckpointedDrm[K] =
// We are already checkpointed in a sense that we already have Spark lineage. So just return self.
this
@@ -95,7 +94,7 @@ class CheckpointedDrmBase[K : ClassTag](
*/
def collect: Matrix = {
- val intRowIndices = implicitly[ClassManifest[K]] == classManifest[Int]
+ val intRowIndices = implicitly[ClassTag[K]] == implicitly[ClassTag[Int]]
val cols = rdd.map(_._2.length).fold(0)(max(_, _))
val rows = if (intRowIndices) rdd.map(_._1.asInstanceOf[Int]).fold(-1)(max(_, _)) + 1 else rdd.count().toInt
@@ -134,7 +133,18 @@ class CheckpointedDrmBase[K : ClassTag](
* Dump matrix as computed Mahout's DRM into specified (HD)FS path
* @param path
*/
- def writeDRM(path: String)(implicit ev: K => Writable) = rdd.saveAsSequenceFile(path)
+ def writeDRM(path: String) = {
+ val ktag = implicitly[ClassTag[K]]
+
+ implicit val k2wFunc: (K) => Writable =
+ if (ktag.runtimeClass == classOf[Int]) (x: K) => new IntWritable(x.asInstanceOf[Int])
+ else if (ktag.runtimeClass == classOf[String]) (x: K) => new Text(x.asInstanceOf[String])
+ else if (ktag.runtimeClass == classOf[Long]) (x: K) => new LongWritable(x.asInstanceOf[Long])
+ else if (classOf[Writable].isAssignableFrom(ktag.runtimeClass)) (x: K) => x.asInstanceOf[Writable]
+ else throw new IllegalArgumentException("Do not know how to convert class tag %s to Writable.".format(ktag))
+// implicit def any2w(k: Any): Writable = k2wFunc(k)
+ rdd.saveAsSequenceFile(path)
+ }
protected def computeNRow = {
Modified: mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/package.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/package.scala?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/package.scala (original)
+++ mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/drm/package.scala Tue Apr 22 18:39:30 2014
@@ -88,7 +88,7 @@ package object drm {
*
* @return DRM[Any] where Any is automatically translated to value type
*/
- def drmFromHDFS(path: String)(implicit sc: SparkContext): CheckpointedDrmBase[Any] = {
+ def drmFromHDFS (path: String)(implicit sc: SparkContext): CheckpointedDrmBase[_] = {
val rdd = sc.sequenceFile(path, classOf[Writable], classOf[VectorWritable]).map(t => (t._1, t._2.get()))
val key = rdd.map(_._1).take(1)(0)
@@ -108,10 +108,16 @@ package object drm {
case xx: Writable => (x: Any) => x.asInstanceOf[Writable]
}
+ val km = key match {
+ case xx: IntWritable => implicitly[ClassTag[Int]]
+ case xx: Text => implicitly[ClassTag[String]]
+ case xx: LongWritable => implicitly[ClassTag[Long]]
+ case xx: Writable => ClassTag(classOf[Writable])
+ }
+
{
- implicit val km: ClassManifest[AnyRef] = ClassManifest.classType(key.getClass)
implicit def getWritable(x: Any): Writable = val2key()
- new CheckpointedDrmBase[Any](rdd.map(t => (key2val(t._1), t._2)))
+ new CheckpointedDrmBase(rdd.map(t => (key2val(t._1), t._2)))(km.asInstanceOf[ClassTag[Any]])
}
}
Modified: mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/package.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/package.scala?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/package.scala (original)
+++ mahout/trunk/spark/src/main/scala/org/apache/mahout/sparkbindings/package.scala Tue Apr 22 18:39:30 2014
@@ -27,9 +27,6 @@ import org.apache.mahout.common.IOUtils
import org.apache.log4j.Logger
import org.apache.mahout.sparkbindings.drm.DrmLike
-/**
- * @author dmitriy
- */
package object sparkbindings {
private[sparkbindings] val log = Logger.getLogger("org.apache.mahout.sparkbindings")
@@ -42,15 +39,15 @@ package object sparkbindings {
* @return
*/
def mahoutSparkContext(masterUrl: String, appName: String,
- customJars: TraversableOnce[String] = Nil): SparkContext = {
+ customJars: TraversableOnce[String] = Nil,
+ sparkConf: SparkConf = new SparkConf(),
+ addMahoutJars: Boolean = true
+ ): SparkContext = {
val closeables = new java.util.ArrayDeque[Closeable]()
try {
- val conf = if (!masterUrl.startsWith("local")
- || System.getProperties.contains("mahout.home")
- || System.getenv("MAHOUT_HOME") != null
- ) {
+ if (addMahoutJars) {
var mhome = System.getenv("MAHOUT_HOME")
if (mhome == null) mhome = System.getProperty("mahout.home")
@@ -65,7 +62,7 @@ package object sparkbindings {
if (!exec.canExecute)
throw new IllegalArgumentException("Cannot execute %s.".format(exec.getAbsolutePath))
- val p = Runtime.getRuntime.exec(Array(exec.getAbsolutePath, "classpath"))
+ val p = Runtime.getRuntime.exec(Array(exec.getAbsolutePath, "-spark", "classpath"))
closeables.addFirst(new Closeable {
def close() {
@@ -105,23 +102,30 @@ package object sparkbindings {
j.matches(".*mahout-math-scala-.*\\.jar") ||
j.matches(".*mahout-mrlegacy-.*\\.jar") ||
j.matches(".*mahout-spark-.*\\.jar")
- ).filter(!_.matches(".*-tests.jar")) ++
- SparkContext.jarOfClass(classOf[DrmLike[_]]) ++ customJars
+ ).filter(n => !n.matches(".*-tests.jar") && !n.matches(".*-sources.jar")) ++
+ SparkContext.jarOfClass(classOf[DrmLike[_]])
if (log.isDebugEnabled) {
log.debug("Mahout jars:")
mcjars.foreach(j => log.debug(j))
}
- new SparkConf().setJars(jars = mcjars.toSeq)
+ sparkConf.setJars(jars = mcjars.toSeq ++ customJars)
- } else new SparkConf()
+ } else {
+ // In local mode we don't care about jars, do we?
+ sparkConf.setJars(customJars.toSeq)
+ }
- conf.setAppName(appName).setMaster(masterUrl)
+ sparkConf.setAppName(appName).setMaster(masterUrl)
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.set("spark.kryo.registrator", "org.apache.mahout.sparkbindings.io.MahoutKryoRegistrator")
- new SparkContext(config = conf)
+ if (System.getenv("SPARK_HOME") != null) {
+ sparkConf.setSparkHome(System.getenv("SPARK_HOME"))
+ }
+
+ new SparkContext(config = sparkConf)
} finally {
IOUtils.close(closeables)
Modified: mahout/trunk/spark/src/test/scala/org/apache/mahout/sparkbindings/test/MahoutLocalContext.scala
URL: http://svn.apache.org/viewvc/mahout/trunk/spark/src/test/scala/org/apache/mahout/sparkbindings/test/MahoutLocalContext.scala?rev=1589246&r1=1589245&r2=1589246&view=diff
==============================================================================
--- mahout/trunk/spark/src/test/scala/org/apache/mahout/sparkbindings/test/MahoutLocalContext.scala (original)
+++ mahout/trunk/spark/src/test/scala/org/apache/mahout/sparkbindings/test/MahoutLocalContext.scala Tue Apr 22 18:39:30 2014
@@ -1,7 +1,7 @@
package org.apache.mahout.sparkbindings.test
import org.scalatest.Suite
-import org.apache.spark.SparkContext
+import org.apache.spark.{SparkConf, SparkContext}
import org.apache.mahout.sparkbindings._
import org.apache.mahout.test.MahoutSuite
@@ -13,17 +13,16 @@ trait MahoutLocalContext extends MahoutS
protected implicit var mahoutCtx: SparkContext = _
- // Additional jars?
- protected val buildJars = Traversable.empty[String]
-
override protected def beforeEach() {
super.beforeEach()
- System.setProperty("spark.kryoserializer.buffer.mb","15")
- System.setProperty("spark.akka.frameSize","30")
mahoutCtx = mahoutSparkContext(masterUrl = "local[3]",
appName = "MahoutLocalContext",
- customJars = buildJars
+ // Do not run MAHOUT_HOME jars in unit tests.
+ addMahoutJars = false,
+ sparkConf = new SparkConf()
+ .set("spark.kryoserializer.buffer.mb", "15")
+ .set("spark.akka.frameSize", "30")
)
}