Showing posts with label Mahout. Show all posts
Showing posts with label Mahout. Show all posts

Friday, September 9, 2011

GraphLab Clustering library

Recently I have been working on implementing a clustering library on top of GraphLab.
Currently we have K-means, Fuzzy K-means and LDA (Latent Dirichlet Allocation) implemented. I took some time for comparing performance of GraphLab vs. Mahout on an Amazon EC2 machine.

Here is a graph which compares performance:


Some explanation about the experiment. I took a subset of Netflix data with 3,298,163 movie ratings, 95,526 users, and 3,561 movies. The goal is to cluster user with similar movie preferences together. Both GraphLab and Mahout run on Amazon m2.xlarge instance .
This machine has 2 cores. I have used the following settings: 250 clusters, 50 clusters and 20 clusters. The algorithm runs a single iteration and then dumps the output into a text file.
For Mahout, I used Mahout's K-Means implementation. GraphLab was run using a single node, while Mahout was run using either one or two nodes. Mahout is using 7 mappers and GraphLab 7 threads.

Overall, GraphLab runs between x15 to x40 faster on this dataset.

A second experiment I did is to compare Mahout's LDA performance to GraphLab's LDA.
Here is the Graph:
For this experiment, I used m1.xlarge instance. I tested Graphlab on 4 cores, Mahout and 4 cores and Mahout on 8 cores (2 nodes). I used the same Netflix data subset, this time with 10 clusters. Graph depicts running time of a single iteration.

Finally, here are performance results of GraphLab LDA with 1, 2, 3 and 4 cores (on m1.xlarge EC2 instance):

Running time in this case is for 5 iterations.

Saturday, September 3, 2011

Understanding Mahout K-Means clustering implementation

This post helps to understand Mahout's K-Means clustering implementation.
Preliminaries: you should read first the explanation in the link above.

Installation and setup
wget http://apache.spd.co.il//mahout/0.5/mahout-distribution-0.5.zip
unzip mahout-distribution-0.5.zip
cd mahout-distribution-0.5.zip 
setenv JAVA_HOME /path.to/java1.6.0/

Running the example
From the Mahout root folder:
./examples/bin/build_reuters.sh

Explanation 
The script build_reuters.sh downloads reuters data, which is composed of news items.
<46|0>bickson@biggerbro:~/usr7/mahout-distribution-0.5/examples/bin/mahout-work/reuters-out$ ls
reut2-000.sgm-0.txt    reut2-003.sgm-175.txt  reut2-006.sgm-24.txt   reut2-009.sgm-324.txt  reut2-012.sgm-39.txt   reut2-015.sgm-474.txt  reut2-018.sgm-549.txt
reut2-000.sgm-100.txt  reut2-003.sgm-176.txt  reut2-006.sgm-250.txt  reut2-009.sgm-325.txt  reut2-012.sgm-3.txt    reut2-015.sgm-475.txt  reut2-018.sgm-54.txt
....
A typical news item looks like:
26-FEB-1987 15:01:01.79

BAHIA COCOA REVIEW

Showers continued throughout the week in the Bahia cocoa zone, alleviating the drought since early January and improving prospects for the coming temporao, although normal humidity levels have not been restored, Comissaria Smith said in its weekly review.     The dry period means the temporao will be late this year.     Arrivals for the week ended February 22 were 155,221 bags of 60 kilos making a cumulative total for the season of 5.93 mln against 5.81 at the same stage last year. Again it seems th
....

The goal of the method, is to cluster similar news items together. This is done by first counting word occurrences using TF-IDF scheme. Each news item is a sparse row in a matrix. Next, rows are clustered together using the k-means algorithm. 

What happens behind the scenes?
1) mahout seqdirectory is called, to create sequence files containing file name as key, and file content as value.
INPUT DIR: mahout-work/reuters-out/
OUTPUT DIR: mahout-work/reuters-out-seqdir/

2) mahout seq2parse is called, to create sparse vectors out of the sequence files.
INPUT DIR: mahout-work/reuters-out-seqdir/
OUTPUT DIR: mahout-work/reuters-out-seqdir-sparse-kmeans/tfidf-vectors/

Inside the output dir, a file called part-t-00000 is created. This is a sequence file which includes int (row id) as key, and a sparse vector (SequentialAccessSparseVector) as value.

3) mahout kmeams is called, for clustering the sparse vectors into cluster.
INPUT DIR: mahout-work/reuters-out-seqdir-sparse-kmeans/tfidf-vectors/
INTERMEDIATE OUTPUT DIR: mahout-work/reuters-kmeans-clusters/
OUTPUT DIR:mahout-work/reuters-kmeans/

4) Finally clusterdump converts clusters into human readable format
INPUT DIR: mahout-work/reuters-kmeans/
OUPUT : a text file.

Debugging:
Below you can find some common problems and their solutions.

Problem:
~/usr7/mahout-distribution-0.5$ ./bin/mahout kmeans -i ~/usr7/small_netflix_mahout/ -o ~/usr7/small_netflix_mahout_output/ --numClusters 10 -c ~/usr7/small_netflix_mahout/ -x 10
no HADOOP_HOME set, running locally
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/mahout-examples-0.5-job.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/lib/slf4j-jcl-1.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
Sep 4, 2011 2:10:39 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Command line arguments: {--clusters=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/, --convergenceDelta=0.5, --distanceMeasure=org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure, --endPhase=2147483647, --input=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/, --maxIter=10, --method=mapreduce, --numClusters=10, --output=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_output/, --startPhase=0, --tempDir=temp}
Sep 4, 2011 2:10:39 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Deleting /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout
Sep 4, 2011 2:10:39 AM org.apache.hadoop.util.NativeCodeLoader <clinit>
WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Sep 4, 2011 2:10:39 AM org.apache.hadoop.io.compress.CodecPool getCompressor
INFO: Got brand-new compressor
Sep 4, 2011 2:10:39 AM org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker <init>
WARNING: Problem opening checksum file: file:/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/part-randomSeed.  Ignoring exception: java.io.EOFException
    at java.io.DataInputStream.readFully(DataInputStream.java:180)
    at java.io.DataInputStream.readFully(DataInputStream.java:152)
    at org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker.<init>(ChecksumFileSystem.java:134)
    at org.apache.hadoop.fs.ChecksumFileSystem.open(ChecksumFileSystem.java:283)
    at org.apache.hadoop.io.SequenceFile$Reader.openFile(SequenceFile.java:1437)
    at org.apache.hadoop.io.SequenceFile$Reader.<init>(SequenceFile.java:1424)
    at org.apache.hadoop.io.SequenceFile$Reader.<init>(SequenceFile.java:1417)
    at org.apache.hadoop.io.SequenceFile$Reader.<init>(SequenceFile.java:1412)
    at org.apache.mahout.common.iterator.sequencefile.SequenceFileIterator.<init>(SequenceFileIterator.java:58)
    at org.apache.mahout.common.iterator.sequencefile.SequenceFileIterable.iterator(SequenceFileIterable.java:61)
    at org.apache.mahout.clustering.kmeans.RandomSeedGenerator.buildRandom(RandomSeedGenerator.java:87)
    at org.apache.mahout.clustering.kmeans.KMeansDriver.run(KMeansDriver.java:101)
    at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
    at org.apache.mahout.clustering.kmeans.KMeansDriver.main(KMeansDriver.java:58)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.ProgramDriver$ProgramDescription.invoke(ProgramDriver.java:68)
    at org.apache.hadoop.util.ProgramDriver.driver(ProgramDriver.java:139)
    at org.apache.mahout.driver.MahoutDriver.main(MahoutDriver.java:187)

Answer: cluster path and input path point for the same folder. When starting run all files in cluster path are deleted, so input file is deleted as well. Change paths to point to different folders! Problem:
./bin/mahout kmeans -i ~/usr7/small_netflix_mahout/ -o ~/usr7/small_netflix_mahout_output/ --numClusters 10 -c ~/usr7/small_netflix_mahout_clusters/ -x 10
no HADOOP_HOME set, running locally
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/mahout-examples-0.5-job.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/lib/slf4j-jcl-1.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
Sep 4, 2011 2:15:11 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Command line arguments: {--clusters=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters/, --convergenceDelta=0.5, --distanceMeasure=org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure, --endPhase=2147483647, --input=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/, --maxIter=10, --method=mapreduce, --numClusters=10, --output=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_output/, --startPhase=0, --tempDir=temp}
Sep 4, 2011 2:15:12 AM org.apache.hadoop.util.NativeCodeLoader <clinit>
WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Sep 4, 2011 2:15:12 AM org.apache.hadoop.io.compress.CodecPool getCompressor
INFO: Got brand-new compressor
Exception in thread "main" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
    at java.util.ArrayList.RangeCheck(ArrayList.java:547)
    at java.util.ArrayList.get(ArrayList.java:322)
    at org.apache.mahout.clustering.kmeans.RandomSeedGenerator.buildRandom(RandomSeedGenerator.java:108)
    at org.apache.mahout.clustering.kmeans.KMeansDriver.run(KMeansDriver.java:101)
    at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
    at org.apache.mahout.clustering.kmeans.KMeansDriver.main(KMeansDriver.java:58)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
    at java.lang.reflect.Method.invoke(Method.java:597)
    at org.apache.hadoop.util.ProgramDriver$ProgramDescription.invoke(ProgramDriver.java:68)
    at org.apache.hadoop.util.ProgramDriver.driver(ProgramDriver.java:139)
    at org.apache.mahout.driver.MahoutDriver.main(MahoutDriver.java:187)

Answer: Input file named part-r-00000 is missing in the input folder. Sucessful run:
124|0>bickson@biggerbro:~/usr7/mahout-distribution-0.5$ ./bin/mahout kmeans -i ~/usr7/small_netflix_mahout/ -o ~/usr7/small_netflix_mahout_output/ --numClusters 10 -c ~/usr7/small_netflix_mahout_clusters/ -x 10
no HADOOP_HOME set, running locally
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/mahout-examples-0.5-job.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/lib/slf4j-jcl-1.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
Sep 4, 2011 2:19:48 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Command line arguments: {--clusters=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters/, --convergenceDelta=0.5, --distanceMeasure=org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure, --endPhase=2147483647, --input=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/, --maxIter=10, --method=mapreduce, --numClusters=10, --output=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_output/, --startPhase=0, --tempDir=temp}
Sep 4, 2011 2:19:48 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Deleting /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters
Sep 4, 2011 2:19:48 AM org.apache.hadoop.util.NativeCodeLoader <clinit>
WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Sep 4, 2011 2:19:48 AM org.apache.hadoop.io.compress.CodecPool getCompressor
INFO: Got brand-new compressor
Sep 4, 2011 2:19:48 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 2:19:48 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 2:19:48 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 2:19:48 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 2:19:52 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Wrote 10 vectors to /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters/part-randomSeed
Sep 4, 2011 2:19:52 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Input: /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout Clusters In: /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters/part-randomSeed Out: /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_output Distance: org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure
Sep 4, 2011 2:19:52 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: convergence: 0.5 max Iterations: 10 num Reduce Tasks: org.apache.mahout.math.VectorWritable Input Vectors: {}
Sep 4, 2011 2:19:52 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: K-Means Iteration 1
Sep 4, 2011 2:19:52 AM org.apache.hadoop.metrics.jvm.JvmMetrics init
INFO: Initializing JVM Metrics with processName=JobTracker, sessionId=
Sep 4, 2011 2:19:52 AM org.apache.hadoop.mapreduce.lib.input.FileInputFormat listStatus
INFO: Total input paths to process : 1
Sep 4, 2011 2:19:53 AM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
INFO: Running job: job_local_0001
Sep 4, 2011 2:19:53 AM org.apache.hadoop.mapreduce.lib.input.FileInputFormat listStatus
INFO: Total input paths to process : 1
Sep 4, 2011 2:19:53 AM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
INFO: io.sort.mb = 100
Sep 4, 2011 2:19:53 AM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
INFO: data buffer = 79691776/99614720
Sep 4, 2011 2:19:53 AM org.apache.hadoop.mapred.MapTask$MapOutputBuffer <init>
INFO: record buffer = 262144/327680
Sep 4, 2011 2:19:53 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 2:19:54 AM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
INFO:  map 0% reduce 0%
Sep 4, 2011 2:19:59 AM org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
INFO: 
Sep 4, 2011 2:20:00 AM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
INFO:  map 80% reduce 0%
Sep 4, 2011 2:20:00 AM org.apache.hadoop.mapred.MapTask$MapOutputBuffer flush
INFO: Starting flush of map output
Sep 4, 2011 2:20:02 AM org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
INFO: 
Sep 4, 2011 2:20:03 AM org.apache.hadoop.mapred.JobClient monitorAndPrintJob
INFO:  map 100% reduce 0%
Sep 4, 2011 2:20:05 AM org.apache.hadoop.mapred.LocalJobRunner$Job statusUpdate
INFO: 
Problem:
no HADOOP_HOME set, running locally
Exception in thread "main" java.lang.ClassFormatError: org.apache.mahout.driver.MahoutDriver (unrecognized class file version)
   at java.lang.VMClassLoader.defineClass(libgcj.so.8rh)
   at java.lang.ClassLoader.defineClass(libgcj.so.8rh)
   at java.security.SecureClassLoader.defineClass(libgcj.so.8rh)
   at java.net.URLClassLoader.findClass(libgcj.so.8rh)
   at java.lang.ClassLoader.loadClass(libgcj.so.8rh)
   at java.lang.ClassLoader.loadClass(libgcj.so.8rh)
   at gnu.java.lang.MainThread.run(libgcj.so.8rh)
ANSWER: wrong java version used - you should is 1.6.0 or higher. Problem:
../../bin/mahout: line 201: /usr/share/java-1.6.0//bin/java: No such file or directory
../../bin/mahout: line 201: exec: /usr/share/java-1.6.0//bin/java: cannot execute: No such file or directory
Answer: JAVA_HOME is pointing to the wrong place. Inside this directory a subdirectory called bin should be present, with an executable named "java" in it. Problem:
export JAVA_HOME=/afs/cs.cmu.edu/local/java/amd64_f7/jdk1.6.0_16/
cd ~/usr7/mahout-distribution-0.5/ ; ./bin/mahout clusterdump --seqFileDir ~/usr7/small_netflix_mahout_clusters/ --pointsDir ~/usr7/small_netflix_mahout/ --output small_netflix_output.txt
no HADOOP_HOME set, running locally
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/mahout-examples-0.5-job.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/lib/slf4j-jcl-1.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
Sep 4, 2011 4:22:29 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Command line arguments: {--dictionaryType=text, --endPhase=2147483647, --output=small_netflix_output.txt, --pointsDir=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout/, --seqFileDir=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_clusters/, --startPhase=0, --tempDir=temp}
Sep 4, 2011 4:22:29 AM org.apache.hadoop.util.NativeCodeLoader 
WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Sep 4, 2011 4:22:29 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 4:22:29 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 4:22:29 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Sep 4, 2011 4:22:29 AM org.apache.hadoop.io.compress.CodecPool getDecompressor
INFO: Got brand-new decompressor
Exception in thread "main" java.lang.ClassCastException: org.apache.mahout.math.VectorWritable cannot be cast to org.apache.mahout.clustering.WeightedVectorWritable
	at org.apache.mahout.utils.clustering.ClusterDumper.printClusters(ClusterDumper.java:171)
	at org.apache.mahout.utils.clustering.ClusterDumper.run(ClusterDumper.java:121)
	at org.apache.mahout.utils.clustering.ClusterDumper.main(ClusterDumper.java:86)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at org.apache.hadoop.util.ProgramDriver$ProgramDescription.invoke(ProgramDriver.java:68)
	at org.apache.hadoop.util.ProgramDriver.driver(ProgramDriver.java:139)
	at org.apache.mahout.driver.MahoutDriver.main(MahoutDriver.java:187)
make: *** [clusterdump] Error 1
Problem:
export JAVA_HOME=/afs/cs.cmu.edu/local/java/amd64_f7/jdk1.6.0_16/
cd ~/usr7/mahout-distribution-0.5/ ; ./bin/mahout kmeans -i ~/usr7/small_netflix_transpose_mahout/ -o ~/usr7/small_netflix_mahout_transpose_output/ --numClusters 10 -c ~/usr7/small_netflix_mahout_transpose_clusters/ -x 2 -ow -cl
no HADOOP_HOME set, running locally
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/mahout-examples-0.5-job.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/mnt/bigbrofs/usr7/bickson/mahout-distribution-0.5/lib/slf4j-jcl-1.6.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
Sep 4, 2011 4:57:44 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Command line arguments: {--clustering=null, --clusters=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_transpose_clusters/, --convergenceDelta=0.5, --distanceMeasure=org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure, --endPhase=2147483647, --input=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_transpose_mahout/, --maxIter=2, --method=mapreduce, --numClusters=10, --output=/mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_transpose_output/, --overwrite=null, --startPhase=0, --tempDir=temp}
Sep 4, 2011 4:57:45 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Deleting /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_transpose_output
Sep 4, 2011 4:57:45 AM org.slf4j.impl.JCLLoggerAdapter info
INFO: Deleting /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_mahout_transpose_clusters
Exception in thread "main" java.io.FileNotFoundException: File /mnt/bigbrofs/usr6/bickson/usr7/small_netflix_transpose_mahout does not exist.
	at org.apache.hadoop.fs.RawLocalFileSystem.getFileStatus(RawLocalFileSystem.java:361)
	at org.apache.hadoop.fs.FilterFileSystem.getFileStatus(FilterFileSystem.java:245)
	at org.apache.mahout.clustering.kmeans.RandomSeedGenerator.buildRandom(RandomSeedGenerator.java:69)
	at org.apache.mahout.clustering.kmeans.KMeansDriver.run(KMeansDriver.java:101)
	at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
	at org.apache.mahout.clustering.kmeans.KMeansDriver.main(KMeansDriver.java:58)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:597)
	at org.apache.hadoop.util.ProgramDriver$ProgramDescription.invoke(ProgramDriver.java:68)
	at org.apache.hadoop.util.ProgramDriver.driver(ProgramDriver.java:139)
	at org.apache.mahout.driver.MahoutDriver.main(MahoutDriver.java:187)
make: *** [kmeans_transpose] Error 1
Answer: input directory does not exist.

Problem: program clusterdump runs, with empty txt file as output.
Solution: You probably gave the intermediate cluster path of k-means instead of the output path dir. In this case, program runs and terminates without an error.

Sunday, August 28, 2011

GraphLab and Mahout Compatability - Mahout SVD input format

In the last couple of months, we are working on increasing compatibility of GraphLab large scale machine learning project, with Apache Mahout large scale machine learning project. About 3 months ago, as a first step, we decided to change GraphLab license to Apache license to allow for better interaction between the projects.

This week I wrote some Java code, to allow translation of Mahout Hadoop sequence files using in Mahout's SVD solver into GraphLab's format and back. This will allow users of Mahout that rely on Hadoop infrastructure, to pre-process the matrix factorization problem as before, but change the actual solver used from Mahout SVD to GraphLab. The advantage, is that for problems which fit into memory, GraphLab run about x50 faster.

Here is the Java code:
import java.io.*;
import java.util.Iterator;

import org.apache.mahout.math.SequentialAccessSparseVector;
import org.apache.mahout.math.Vector;
import org.apache.mahout.math.VectorWritable;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;

/**
 * Code for converting Mahout SequenceFile containing SequentialAccessSparseVector to GraphLab format
 * @author bickson
 *
 */

public class SeqFile2GraphLab {


   public static int Cardinality;

        /**
         * 
         * @param args[0] - input svd file
         * @param args[2] - output csv file
         */

  static void writeFloat(DataOutputStream dos, float a) throws java.io.IOException{
    int floatBits = Float.floatToIntBits(a);
    byte floatBytes[] = new byte[4];
    floatBytes[3] = (byte)(floatBits >> 24);
    floatBytes[2] = (byte)(floatBits >> 16);
    floatBytes[1] = (byte)(floatBits >> 8);
    floatBytes[0] = (byte)(floatBits);
    dos.write(floatBytes);
  }
  static void writeInt(DataOutputStream dos, int a) throws java.io.IOException{
    byte floatBytes[] = new byte[4];
    floatBytes[3] = (byte)(a >> 24);
    floatBytes[2] = (byte)(a >> 16);
    floatBytes[1] = (byte)(a >> 8);
    floatBytes[0] = (byte)(a);
    dos.write(floatBytes);
  }



  public static void main(String[] args){

    if (args.length < 6){
      System.err.println("Usage: java SeqFile2GraphLab [input seq file name] [output graphlab file name] [M - number of users] [N - number of movies] [K - number of time bins ] [ e - number of edges] [ OPTIONAL: transpose = false]");
      System.exit(1);
    }

    String inputfile = args[0];
    String outputfile = args[1];
   int M = Integer.parseInt(args[2]);
    int N = Integer.parseInt(args[3]);
    int K = Integer.parseInt(args[4]);
    int e = Integer.parseInt(args[5]);
    if (M < 1 || N < 1 || K< 1 || e<1){
       System.err.println("wrong input. M,N,K,e should be >=1");
       System.exit(1);
    }
    boolean transpose = false;
    if (args.length >= 7)
       transpose = new Boolean(args[6]).booleanValue();

    try {
        final Configuration conf = new Configuration();
        final FileSystem fs = FileSystem.get(conf);
        final SequenceFile.Reader reader = new SequenceFile.Reader(fs, new Path(inputfile), conf);
        DataOutputStream dos = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(outputfile)));
        writeInt(dos,M);
        writeInt(dos,N);
        writeInt(dos,K);
        writeInt(dos,e);
        System.out.println("Writing a matrix of size: " + M + " users, " + N + " movies, " + K + " time bins " + e + " edges.");
        IntWritable key = new IntWritable();
        VectorWritable vec = new VectorWritable();
        int edges = 0;

        while (reader.next(key, vec)) {
           SequentialAccessSparseVector vect = (SequentialAccessSparseVector)vec.get();
           //System.out.println("key " + key + " value: " + vect);
           Iterator<Vector.Element> iter = vect.iterateNonZero();

           while(iter.hasNext()){

              Vector.Element element = iter.next();
              if (!transpose){
                assert(element.index() < M);
                assert(key.get() < N);
                writeFloat(dos,element.index()+1);
                writeFloat(dos,key.get()+1+M);
              }
              else {
                assert(element.index() < N);
                assert(key.get() < M);
                writeFloat(dos,key.get()+1);
                writeFloat(dos,element.index()+1+M);
              }
              writeFloat(dos,1);
              writeFloat(dos,(float)vect.getQuick(element.index()));
              //System.out.println("col: " + key.get() + " row: " + element.index() + " val: " + vect.getQuick(element.index()));
              edges++;

              if (edges % 1000000 == 0)
                 System.out.println("edges: " + edges);
           }
        }
        if (edges != e){
           System.err.println("Wrong number of edges in file. Should be " + e + " in practice we have : " + edges);
        }
        reader.close();
        dos.close();
        System.out.println("Done writing !" + e + " edges");
    } catch(Exception ex){
       ex.printStackTrace();
    }

  }
}
Compilation:
javac -cp /mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/core-3.1.1.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-core-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-math-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-cli-1.2.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/hadoop-0.20.2-core.jar *.java
Example run, for converting netflix data, from Mahout's format into GraphLab's:
run2:
        java -cp .:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/core-3.1.1.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-core-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-math-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-cli-1.2.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/hadoop-0.20.2-core.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-logging-1.0.4.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-logging-api-1.0.4.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/commons-collections-3.2.1.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/google-collections-1.0-rc2.jar  SeqFile2GraphLab netflix.seq netflix 480189 17770 27 1408395
        

Thursday, June 9, 2011

What are the most widely deployed machine learning algorithms?

One of the interesting questions is: "what are the most useful machine learning algorithms?".
I did a little survey by looking at the Mahout user mailing list and counting occurrences of keywords. The results I got are shown in the plot above.


It seems that matrix factorization (SVD) is the most widely used algorithm, and then K-means. We have just implemented SVD as a part of the GraphLab Collaborative Filtering library. Anyone who wants to beta test it is welcome!

Friday, March 4, 2011

Tuning Hadoop configuration for high performance - Mahut on Amazon EC2

In this post I will share some of the insights I got when tuning Hadoop/Mahout on Amazon EC regular and high performance nodes. I was using two algorithms.
1) Mahout's Alternating least squares application (See MAHOUT-542) with Netflix data. (Sparse matrix with 100,000,000 non zeros). Test was done with up to 64 HPC nodes (512 cores).
2) CoEM algorithm - NLP algorithm (R. Jones, 2005) with data graph of around 200,000,000 edges.

Below are running time results for running one iteration of alternating least squares (implemented by Sebastian Schelter) on Netflix data. Runtime is in seconds.
X-axis are the participating machines - from 4 to 64 machines.











My conclusion from this experiment, is that 16 HPC nodes (256 cores) are enough for computing matrix factorization/CoEM of this scale. Beyond 16 nodes there is no benefit in further parallism.

Below I explain how I fine-tuned performance.
Preliminaries: I assume you followed the instruction on  part 1 of this tutorial to setup Hadoop on EC2.


1) The hdfs-site.xml file
dfs.replication
- I set dfs replication to 1. Replication determines the number of copies the hdfs data is saved on. When working with a relative low number of nodes (several) higher replication delays performance.

hadoop.tmp.dir
hadoop.data.dir
dfs.name.dir
You should set all those directories to point to DIFFERENT paths which have ENOUGH DISK SPACE.
Default hadoop configuration points to either /tmp or /usr/local/hadoop-0.20.2/ and in Amazon
EC2 there is a 10Gb disk space limit for the root partition. To increase available storage,
on regular nodes I set the above fields to /mnt/tmp1, /mnt/tmp2/ and /mnt/tmp3
On HPC nodes, I first mounted /dev/sdb using the command:
mkdir -p /home/data
mount -t ext3 /dev/sdb/ /home/data/
And then created /home/data/tmp1 /home/data/tmp2 /home/data/tmp3 and pointed the above fields to there.

dfs.block.size
The default is 64MB. For CoEM set it to 4MB, so there will be enough mappers for all cores. For Netflix data I set it to 16MB. When the block size is too small, there are too manny mappers, resulting in loading the system, having many task failures, and some of the job trackers gets black-listed. Having too few mappers does not exploit well parallism. Unfortunately it seems that block size should be tuned separately for each algorithm.

2) The file core-site.xml should be configured as explained in the first part of this post.

3) The file mapred-site.xml
mapred.map.task
empirically setting them to the number of  cores -1 seemed to work the best. (On HPC nodes, 15 cores). Note that this number is per machine.
mapred.reduce.task
Common practice says to set it to 0.95 * number of machines * (number of cores-1).
For me that did not work well, especially with 64 machines - reduce phase becomes terribly slow with very slow copying phase (in Kb instead of MB). Finally I set it to 64 for all experiments.

mapred.tasktracker.map.tasks.maximum, mapred.tasktracker.reduce.tasks.maximum
set them to the values above. Note that it seems that reduce tasks maximum is a global maximum and not a limit per single machines. So in this case 64 was a global limit of 64 reduce tasks.

mapred.task.timeout, mapred.tasktracker.expiry.interval
default is 600000 milliseconds which was too low for ALS. If the interval is too low, task will be killed prematurely. I set it to 7200000
mapred.task.tracker.expiry.interval
don't ask me what is the difference to previous field - probably a bug. Anyway I set it as well.

mapred.compress.map.output, mapred.output.compress
again I set those fields to true. It reduced
significantly the disk writes to about 1/3 the size.

mapred.child.java.opts
set it to -Xmx2500Mb , the default is 500, which results in out of memory errors, java heap errors and GC errors.


4) The file hadoop-env.sh
On HPC nodes, set
JAVA_HOME=/usr/lib/jvm/jre-openjdk
On regular nodes, set
JAVA_HOME=/usr/lib/jvm/java-6-openjdk
Heap size parameter controls the heap size. When it is too small you get
out of memory error and out of heap size erros.
HADOOP_HEAPSIZE=4000

5) Avoiding string parsing as much as possible
Java string parsing is rather slow. Avoid reading string input files as possible and write the data in binary format whenever possible. For the CoEM algorithm, avoiding string parsing resulted in x4 faster code, since the inputs files where read on each iteration.

Some tips I got from Julio Lopez, OpenCloud project @ CMU:
Block size and controlling the number of mappers. I believe someone already commented on this. In general, you want to have the block sizes relatively large in order to induce your job to perform sequential instead of random I/O. You can use the "InputFormat" to control how the work is split and how many tasks are created.

I've found that the first instincts users have is to match the number of mappers or reducers per node to the number of cores. For many Hadoop applications, this does not work. Properly setting these parameters is application dependent (module the available resources). In Hadoop these are framework-wide parameters. In my experience, how memory is allocated to tasks has a much larger impact on application performance. However, it is not clear how these memory parameters should be set, and there are all sorts of complex interactions among tasks.

For reference, in the cloud cluster, there are 8 cores per node, we allow 10 simultaneous tasks to execute per node and in general we see better throughput that way. As I mentioned earlier, most jobs experience contention for memory.

Interesting related projects/ papers: 
1) http://www.cs.duke.edu/~shivnath/amr.html
2) Kai Ren, Julio López and Garth Gibson. Otus: Resource Attribution in Data-Intensive Clusters. MapReduce: The Second International Workshop on MapReduce and its Applications. San Jose, CA, June 2011. (bib, pdf)


Other useful tips:

When stopping and starting Hadoop you should be very careful since Hadoop generates a zillion of temp file, that if found on the next run makes a mess.

1) I always run from script
echo Y | hadoop namenode -format
Since if the file system was formatted the script will get stuck without getting the "Y" input.

2) Remove all /tmp/*.pid files, or else Hadoop will think some old processes are running.

3) Remove all files in the directories hadoop.tmp.dir, hadoop.data.dir, dfs.name.dir
especially VERSION files. Old VERSION files lead to namespaceID collisions.

4) Delete old logs from /usr/local/hadoop-0.20.2/logs/

Friday, February 25, 2011

Mahout on Amazon EC2 - part 5 - installing Hadoop/Mahout on high performance instance (CentOS/RedHat)

This post explains how to install Mahout ML framework on top of Amazon EC2 (CentOS/RedHat based machine).
The notes are based on older Mahout notes: https://cwiki.apache.org/MAHOUT/mahout-on-amazon-ec2.html which are unfortunately outdated.

Note: part 1 of this post, explains how to install the same installation on top of Ubuntu based machine.

Full procedure should take around 2-3  hours.. :-(

1) Start high performance instance from amazon aws console
Cent OS AMI ID ami-7ea24a17 (x86_64)  Edit AMI
Name:  Basic Cluster Instances HVM CentOS 5.4   
Description:  Minimal CentOS 5.4, 64-bit architecture, and HVM-based virtualization for use with Amazon EC2 Cluster Instances.

2) Login into the instance (right mouse click on running instance from AWS console)

3) Install some required stuff
sudo yum update
sudo yum upgrade
sudo apt-get install python-setuptools  
sudo easy_install "simplejson"

4) Install boto (unfortunately I was not able to install it using easy_install directly)
wget http://boto.googlecode.com/files/boto-1.8d.tar.gz
tar xvzf boto-1.8d.tar.gz
cd boto=1.8d
sudo easy_install .

5) Install maven2 (unfortunately I was not able to install it using yum)
wget http://www.trieuvan.com/apache/maven/binaries/apache-maven-2.2.1-bin.tar.gz
tar xvzf apache-maven-2.2.1-bin.tar.gz
cp -R apache-maven-2.2.1 /usr/local/
ln -s /usr/local/apache-maven-2.2.1/bin/mvn /usr/local/bin/

6) Download and install Hadoop
wget http://apache.cyberuse.com//hadoop/core/hadoop-0.20.2/hadoop-0.20.2.tar.gz   
tar vxzf hadoop-0.20.2.tar.gz  
sudo  mv hadoop-0.20.2 /usr/local/

add the following to $HADOOP_HOME/conf/hadoop-env.sh
export JAVA_HOME=/usr/lib/jvm/jre-openjdk/  
# The maximum amount of heap to use, in MB. Default is 1000  
export HADOOP_HEAPSIZE=2000  

add the following to $HADOOP_HOME/conf/core-site.xml and also $HADOOP_HOME/conf/mapred-site.xml

<configuration>
<property>
<name>fs.default.name</name>
<value>hdfs://localhost:9000</value>
</property>   <property>
<name>mapred.job.tracker</name>  
<value>localhost:9001</value>
</property>
 <property>  
<name>dfs.replication</name>  
 <value>1</value>      
  </property>
</configuration>


Edit the file hdfs-site.xml

<configuration>
 <property>
  <name>hadoop.tmp.dir</name>
  <value>/home/data/tmp/</value>
 </property>
<property>
 <name>dfs.data.dir</name>
 <value>/home/data/tmp2/</value>
</property>
<property>
 <name>dfs.name.dir</name>
 <value>/home/data/tmp3/</value>
</property>
</configuration>

Note: directory /home/data does not exist, and you will have to create it
when starting the instance using the commands:
# mkdir -p /home/data  
# mount -t ext3 /dev/sdb/ /home/data/ 
The reason for this setup is that the root dir has only 10GB, while /dev/sdb/
has 800GB.

set up authorized keys for localhost login w/o passwords and format your name node
# ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa
# cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys

  • Checkout and build Mahout from trunk. Alternatively, you can upload a Mahout release tarball and install it as we did with the Hadoop tarball (Don't forget to update your .profile accordingly).

    # svn co http://svn.apache.org/repos/asf/mahout/trunk mahout
    # cd mahout
    # mvn clean install
    # cd ..
    # sudo mv mahout /usr/local/mahout-0.4
    


    4)Add the following to your .profile
    export JAVA_HOME=/usr/lib/jvm/java-6-openjdk
    export HADOOP_HOME=/usr/local/hadoop-0.20.2
    export HADOOP_CONF_DIR=/usr/local/hadoop-0.20.2/conf
    export MAHOUT_HOME=/usr/local/mahout-0.4/
    export MAHOUT_VERSION=0.4-SNAPSHOT
    export MAVEN_OPTS=-Xmx1024m
    

    Verify that the paths on .profile point to the exact version you downloaded

    6) Run Hadoop, just to prove you can, and test Mahout by building the Reuters dataset on it. Finally, delete the files and shut it down.

    # $HADOOP_HOME/bin/hadoop namenode -format
    $HADOOP_HOME/bin/start-all.sh
    jps     // you should see all 5 Hadoop processes (NameNode, SecondaryNameNode, DataNode, JobTracker, TaskTracker)
    cd $MAHOUT_HOME
    ./examples/bin/build-reuters.sh
    $HADOOP_HOME/bin/stop-all.sh
    rm -rf /tmp/*   // delete the Hadoop files


  • Remove the single-host stuff you added to $HADOOP_HOME/conf/core-site.xml and $HADOOP_HOME/conf/mapred-site.xml in step #6b and verify you are happy with the other conf file settings. The Hadoop startup scripts will not make any changes to them. In particular, upping the Java heap size is required for many of the Mahout jobs.
    // edit $HADOOP_HOME/conf/mapred-site.xml to include the following:
    <property>
       <name>mapred.child.java.opts</name>
       <value>-Xmx2000m</value>
    </property>

    7) Allow for Hadoop to run even if you will work on a different EC2 machine:
    echo "NoHostAuthenticationForLocalhost yes" >>~/.ssh/config
    

    8) Now bundle the image.
    Using Amazon AWS console - select running instance, right mouse click and then bundle EBS image. Enter image name and description. Now the machine will reboot and the image will be created.
  • Thursday, February 24, 2011

    Some thoughts about accuracy of Mahout's SVD

    I was testing Mahout's SVD code and I encountered some subtleties.
    I wonder if I am missing anything or is there a bug in the code?

    1) The ordering of eigenvalues was the opposite than eigenvectors. But this was hopefully fixed by now in patch-369.
    2) When requesting a rank of 4, we get 3 eigenvalues... So it seems that the rank is always lower by one.
    3) There are two transformations which makes comparison of results with matlab (or pen & paper) harder:

    a) The scaleFactor. Defined in: ./math/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosState.java
    I quote a documentation remark in: ./math/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java:48
    " /** To avoid floating point overflow problems which arise in power-methods like Lanczos, an initial pass is made
     * through the input matrix to
    generate a good starting seed vector by summing all the rows of the input matrix, and
    compute the trace(inputMatrixt*matrix)
    This latter value, being the sum of all of the singular values, is used to rescale the entire matrix, effectively forcing the largest singular value to be strictly less than one, and transforming floating point overflow
    problems into floating point underflow (ie, very small singular values will become invisible, as they  will appear to be zero and the algorithm will terminate).*/
    

    b) The second transformation is orthonogolization of the resulting vector. This step is optional (IMHO).
    see: ./math/src/main/java/org/apache/mahout/math/decomposer/lanczos/LanczosSolver.java:118
    The function call is: orthoganalizeAgainstAllButLast(nextVector, state);
    Again I quote from documentation:
    /** 

    This implementation uses {@link org.apache.mahout.math.matrix.linalg.EigenvalueDecomposition} to do the * eigenvalue extraction from the small (desiredRank x desiredRank) tridiagonal matrix. Numerical stability is * achieved via brute-force: re-orthogonalization against all previous eigenvectors is computed after every pass. * This can be made smarter if (when!) this proves to be a major bottleneck. Of course, this step can be parallelized * as well. *


    If anyone wants to reproduce my test, Can can add the function testLanczosSolver2() to TestLanczosSolver.java (code below).
    1) To run it, you need first to comment the line:
    //nextVector.assign(new Scale(1 / scaleFactor));
    in LanczosSolver.java, so it is easier to compare the results to Matlab, without the scaling.
    2) You need to also comment the line:
    //orthoganalizeAgainstAllButLast(nextVector, basis);
    in LanczosSolver.java

    The factorized matrix is:
    >> A
    
    3.1200 -3.1212 -3.0000
    -3.1110 1.5000 2.1212
    -7.0000 -8.0000 -4.0000
    
    The eigenvalues are;
    >> [a,b]=eig(A'*A)
    
    a =
    
    0.2132 -0.8010 -0.5593
    -0.5785 0.3578 -0.7330
    0.7873 0.4799 -0.3871
    
    b =
    
    0.0314 0 0
    0 42.6176 0
    0 0 131.2553
    

    Now I run the unit test testLanczosSolver2 and I get:
    INFO: Lanczos iteration complete - now to diagonalize the tri-diagonal auxiliary matrix.
    Feb 9, 2011 1:25:36 PM org.slf4j.impl.JCLLoggerAdapter info
    INFO: Eigenvector 0 found with eigenvalue 131.25526355941963
    Feb 9, 2011 1:25:36 PM org.slf4j.impl.JCLLoggerAdapter info
    INFO: Eigenvector 1 found with eigenvalue 42.61761063477249
    Feb 9, 2011 1:25:36 PM org.slf4j.impl.JCLLoggerAdapter info
    INFO: Eigenvector 2 found with eigenvalue 0.03137295830779152
    Feb 9, 2011 1:25:36 PM org.slf4j.impl.JCLLoggerAdapter info
    INFO: LanczosSolver finished.
    

    As you can see the eigenvalues are correct.

    @Test
    public void testLanczosSolver2() throws Exception {
    int numRows = 3; int numCols = 3;
    int numColumns = 3;
    SparseRowMatrix m = new SparseRowMatrix(new int[]{numRows, numCols});
    /**
    
        * 3.1200 -3.1212 -3.0000
          -3.1110 1.5000 2.1212
          -7.0000 -8.0000 -4.0000
    
    */
    m.set(0,0,3.12);
    m.set(0,1,-3.12121);
    m.set(0,2,-3);
    m.set(1,0,-3.111);
    m.set(1,1,1.5);
    m.set(1,2,2.12122);
    m.set(2,0,-7);
    m.set(2,1,-8);
    m.set(2,2,-4);
    
    int rank = 4;
    Matrix eigens = new DenseMatrix(rank, numColumns);
    long time = timeLanczos(m, eigens, rank, false);
    assertTrue("Lanczos taking too long! Are you in the debugger? ", time < 10000);
    }
    
    Update: June 1st, 2011: Now GraphLab has also an efficient SVD Lanczos solver. Some performance benchmarks are found here:

    Monday, February 21, 2011

    Large scale matrix factorization using alternating least suqares: which is better - GraphLab or Mahout?

    I am working in the last couple of weeks on comparing the performance of GraphLab vs. Mahout on Alternaring least squares using Netflix data. To remind, GraphLab is the parallel machine learning system we are building in CMU.

    Initial results are encouraging. Mahout Alternating least squares implementation by Sebastian Schelter was tested on Amazon EC2, using two m2.2xlarge nodes (13x2 virtual cores).

    For running 10 iterations, number of features=20, lambda=0.065, it takes 39272 seconds, while GraphLab implementation in C++ takes only 714 seconds (on a machine with 8 cores).

    Running time may be taken with a grain of salt, since I was not using the exact same machine, but the magnitude of difference will certainly hold even if I would run GraphLab on EC2 (which I plan to do soon).

    Regarding accuracy, Mahout ALS has a test RMSE accuracy of
    0.9310 while GraphLab obtained slightly better accuracy of 0.9279.

    Here is Mahout ALS final output: (of the RMSE computation)
    ubuntu@ip-10-115-27-222:/mnt$ /usr/local/mahout-0.4/bin/
    mahout evaluateALS --probes /user/ubuntu/myout/probeSet/ --userFeatures /tmp/als/out/U/ --itemFeatures /tmp/als/out/M/ | grep RMSE
    11/02/17 12:31:42 WARN driver.MahoutDriver: No evaluateALS.props found on classpath, will use command-line arguments only
    11/02/17 12:31:42 INFO common.AbstractJob: Command line arguments: {--endPhase=2147483647, --itemFeatures=/tmp/als/out/M/, --probes=/user/ubuntu/myout/probeSet/, --startPhase=0, --tempDir=temp, --userFeatures=/tmp/als/out/U/}
    RMSE: 0.9310729597725026, MAE: 0.7298745910296568
    11/02/17 12:31:55 INFO driver.MahoutDriver: Program took 12437 ms
    

    Here is the GraphLab output:
    bickson@biggerbro:~/newgraphlab/graphlabapi/debug/apps/pmf$ ./PMF netflix-r 10 0 --D=20 --max_iter=10 --lambda=0.065 --ncpus=8
    setting run mode 0
    INFO   :pmf.cpp(main:1121): PMF starting
    
    loading data file netflix-r
    Loading netflix-r train
    Creating 99072112 edges...
    ................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................loading data file netflix-re
    Loading netflix-re test
    Creating 1408395 edges...
    ........setting regularization weight to 0.065
    PTF_ALS for matrix (480189, 17770, 27):99072112.  D=20
    pU=0.065, pV=0.065, pT=1, muT=1, D=20
    nuAlpha=1, Walpha=1, mu=0, muT=1, nu=20, beta=1, W=1, WT=1 BURN_IN=10
    complete. Obj=6.83664e+08, TEST RMSE=3.7946.
    INFO   :asynchronous_engine.hpp(run:56): Worker 0 started.
    
    ...
    
    INFO   :asynchronous_engine.hpp(run:56): Worker 7 started.
    
    Entering last iter with 1
    228.524) Iter ALS 1  Obj=2.60675e+08, TRAIN RMSE=2.2904 TEST RMSE=0.9948.
    Entering last iter with 2
    289.594) Iter ALS 2  Obj=6.48921e+07, TRAIN RMSE=1.1400 TEST RMSE=0.9573.
    Entering last iter with 3
    350.487) Iter ALS 3  Obj=4.75073e+07, TRAIN RMSE=0.9754 TEST RMSE=0.9444.
    Entering last iter with 4
    411.551) Iter ALS 4  Obj=4.09914e+07, TRAIN RMSE=0.9063 TEST RMSE=0.9381.
    Entering last iter with 5
    472.615) Iter ALS 5  Obj=3.79096e+07, TRAIN RMSE=0.8718 TEST RMSE=0.9348.
    Entering last iter with 6
    533.039) Iter ALS 6  Obj=3.61298e+07, TRAIN RMSE=0.8513 TEST RMSE=0.9324.
    Entering last iter with 7
    594.177) Iter ALS 7  Obj=3.50076e+07, TRAIN RMSE=0.8382 TEST RMSE=0.9305.
    Entering last iter with 8
    654.41) Iter ALS 8  Obj=3.42655e+07, TRAIN RMSE=0.8294 TEST RMSE=0.9290.
    Entering last iter with 9
    714.095) Iter ALS 9  Obj=3.37535e+07, TRAIN RMSE=0.8234 TEST RMSE=0.9279.
    INFO   :asynchronous_engine.hpp(run:66): Worker 6 finished.
    
    ...
    
    INFO   :asynchronous_engine.hpp(run:66): Worker 2 finished.
    

    Wednesday, February 9, 2011

    Hadoop/Mahout - setting up a development environment

    This post explains how to setup a development environment for Hadoop and Mahout.

    Prerequisites:  need to have Mahout and Hadoop sources. (See previous posts).

    On a development machine
    1) Download Helios version of Eclipse like eclipse-java-helios-SR1-linux-gtk-x86_64.tar.gz
    and save it locally. Opem the zip file using:
     tar xvzf *.gz

     2) Install the Map-reduce eclipse plugin
    cd eclipse/plugins/
    wget https://issues.apache.org/jira/secure/attachment/12460491/hadoop-eclipse-plugin-0.20.3-SNAPSHOT.jar

     3) Follow the directions in
    http://m2eclipse.sonatype.org/installing-m2eclipse.html
    to install maven plugin in eclipse.

    4) Eclipse-> File-> import maven project -> select mahout root dir -> finish
    you will see a list of all subprojects. Press OK and wait for compilation to finish.
    If everything went smoothly project should compile.

    5)Select Map-reduce view -> Map-reduce location tab -> Edit Hadoop locations
    In general tab, Add location name (just a name to identify this configuration) and the host and
    port for the Map/reduce master (default port 50030 using the EC2 configuration described in previous posts) and DFS master (default port 50070) -> Finish

    Friday, February 4, 2011

    Mahout - SVD matrix factorization - reading output

    Converting Mahout's SVD Distributed Matrix Factorization Solver Output Format into CSV format

    Purpose
    The code below, shows how to convert a matrix from Mahout's SVD output format
    into a matrix format.


    This code is based on code by Danny Leshem, ContextIn.

    Command line arguments:
    args[0] - path to svd output file
    args[1] - path to output csv file

    Compilation:
    Copy the java code below into an java file named SVD2CSV.java
    Add to the project path both Mahout and Hadoop jars.


    #
    import java.io.BufferedWriter;
    import java.io.FileWriter;
    import java.util.Iterator;
    
    import org.apache.mahout.math.SequentialAccessSparseVector;
    import org.apache.mahout.math.Vector;
    import org.apache.mahout.math.VectorWritable;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.SequenceFile;
    
    
    public class SVD2CSV {
       
       
        public static int Cardinality;
       
        /**
         *
         * @param args[0] - input csv file
         * @param args[1] - cardinality (length of vector)
         * @param args[2] - output file for svd
         */
        public static void main(String[] args){
       
    try {
        final Configuration conf = new Configuration();
        final FileSystem fs = FileSystem.get(conf);
        final SequenceFile.Reader reader = new SequenceFile.Reader(fs, new Path(args[0]), conf);
        BufferedWriter br = new BufferedWriter(new FileWriter(args[1]));
        IntWritable key = new IntWritable();
        VectorWritable vec = new VectorWritable();
             
              while (reader.next(key, vec)) {
                  //System.out.println("key " + key);
                  SequentialAccessSparseVector vect = (SequentialAccessSparseVector)vec.get();
                   System.out.println("key " + key + " value: " + vect);
                  Iterator<Vector.Element> iter = vect.iterateNonZero();
    
                   while(iter.hasNext()){
                      Vector.Element element = iter.next();
                      br.write(key + "," + element.index() + "," + vect.getQuick(element.index())+"\n");
                }
              }
         
              reader.close();
              br.close();
    
           
        } catch(Exception ex){
            ex.printStackTrace();
        }
        }
    }


    When parsing the output please look here about a discussion regarding validity of computed results.

    Further reading: Yahoo! KDD Cup 2011 - large scale matrix factorization.

    Mahout - SVD matrix factorization - formatting input matrix

    Converting Input Format into Mahout's SVD Distributed Matrix Factorization Solver

    Purpose
    The code below, converts a matrix from csv format:
    <from row>,<to col>,<value>\n
    Into Mahout's SVD solver format.


    For example, 
    The 3x3 matrix:
    0    1.0 2.1
    3.0  4.0 5.0
    -5.0 6.2 0


    Will be given as input in a csv file as:
    1,0,3.0
    2,0,-5.0
    0,1,1.0
    1,1,4.0
    2,1,6.2
    0,2,2.1
    1,2,5.0

    NOTE: I ASSUME THE MATRIX IS SORTED BY THE COLUMNS ORDER
    This code is based on code by Danny Leshem, ContextIn.

    Command line arguments:
    args[0] - path to csv input file
    args[1] - cardinality of the matrix (number of columns)
    args[2] - path the resulting Mahout's SVD input file

    Method:
    The code below, goes over the csv file, and for each matrix column, creates a SequentialAccessSparseVector which contains all the non-zero row entries for this column.
    Then it appends the column vector to file.

    Compilation:
    Copy the java code below into an java file named Convert2SVD.java
    Add to your IDE project path both Mahout and Hadoop jars. Alternatively, a command line option for compilation is given below.


    import java.io.BufferedReader;
    import java.io.FileReader;
    import java.util.StringTokenizer;
    
    import org.apache.mahout.math.SequentialAccessSparseVector;
    import org.apache.mahout.math.Vector;
    import org.apache.mahout.math.VectorWritable;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.fs.FileSystem;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.SequenceFile;
    import org.apache.hadoop.io.SequenceFile.CompressionType;
    
    /**
     * Code for converting CSV format to Mahout's SVD format
     * @author Danny Bickson, CMU
     * Note: I ASSUME THE CSV FILE IS SORTED BY THE COLUMN (NAMELY THE SECOND FIELD).
     *
     */
    
    public class Convert2SVD {
    
    
            public static int Cardinality;
    
            /**
             * 
             * @param args[0] - input csv file
             * @param args[1] - cardinality (length of vector)
             * @param args[2] - output file for svd
             */
            public static void main(String[] args){
    
    try {
            Cardinality = Integer.parseInt(args[1]);
            final Configuration conf = new Configuration();
            final FileSystem fs = FileSystem.get(conf);
            final SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, new Path(args[2]), IntWritable.class, VectorWritable.class, CompressionType.BLOCK);
    
              final IntWritable key = new IntWritable();
              final VectorWritable value = new VectorWritable();
    
       
               String thisLine;
            
               BufferedReader br = new BufferedReader(new FileReader(args[0]));
               Vector vector = null;
               int from = -1,to  =-1;
               int last_to = -1;
               float val = 0;
               int total = 0;
               int nnz = 0;
               int e = 0;
               int max_to =0;
               int max_from = 0;
    
               while ((thisLine = br.readLine()) != null) { // while loop begins here
                
                     StringTokenizer st = new StringTokenizer(thisLine, ",");
                     while(st.hasMoreTokens()) {
                         from = Integer.parseInt(st.nextToken())-1; //convert from 1 based to zero based
                         to = Integer.parseInt(st.nextToken())-1; //convert from 1 based to zero basd
                         val = Float.parseFloat(st.nextToken());
                         if (max_from < from) max_from = from;
                         if (max_to < to) max_to = to;
                         if (from < 0 || to < 0 || from > Cardinality || val == 0.0)
                             throw new NumberFormatException("wrong data" + from + " to: " + to + " val: " + val);
                     }
                  
                     //we are working on an existing column, set non-zero rows in it
                     if (last_to != to && last_to != -1){
                         value.set(vector);
                         
                         writer.append(key, value); //write the older vector
                         e+= vector.getNumNondefaultElements();
                     }
                     //a new column is observed, open a new vector for it
                     if (last_to != to){
                         vector = new SequentialAccessSparseVector(Cardinality); 
                         key.set(to); // open a new vector
                         total++;
                     }
    
                     vector.set(from, val);
                     nnz++;
    
                     if (nnz % 1000000 == 0){
                       System.out.println("Col" + total + " nnz: " + nnz);
                     }
                     last_to = to;
    
              } // end while 
    
               value.set(vector);
               writer.append(key,value);//write last row
               e+= vector.getNumNondefaultElements();
               total++;
               
               writer.close();
               System.out.println("Wrote a total of " + total + " cols " + " nnz: " + nnz);
               if (e != nnz)
                    System.err.println("Bug:missing edges! we only got" + e);
              
               System.out.println("Highest column: " + max_to + " highest row: " + max_from );
            } catch(Exception ex){
                    ex.printStackTrace();
            }
        }
    }
    


    A second option to compile this file is create a Makefile, with the following in it:
    all:
            javac -cp /mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/core-3.1.1.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-core-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-math-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-cli-1.2.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/hadoop-0.20.2-core.jar *.java
    
    Note that you will have the change location of the jars to point to where your jars are stored.

    Example for running this conversion for netflix data:
    java -cp .:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/core-3.1.1.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-core-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/mahout-0.4/taste-web/target/mahout-taste-webapp-0.5-SNAPSHOT/WEB-INF/lib/mahout-math-0.5-SNAPSHOT.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-cli-1.2.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/hadoop-0.20.2-core.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-logging-1.0.4.jar:/mnt/bigbrofs/usr7/bickson/hadoop-0.20.2/lib/commons-logging-api-1.0.4.jar Convert2SVD ../../netflixe.csv 17770 netflixe.seq

    Aug 23, 2011 1:16:06 PM org.apache.hadoop.util.NativeCodeLoader
    WARNING: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    Aug 23, 2011 1:16:06 PM org.apache.hadoop.io.compress.CodecPool getCompressor
    INFO: Got brand-new compressor
    Wrote a total of 241 rows, nnz: 1000000
    Wrote a total of 381 rows, nnz: 2000000
    Wrote a total of 571 rows, nnz: 3000000
    Wrote a total of 789 rows, nnz: 4000000
    Wrote a total of 1046 rows, nnz: 5000000
    Wrote a total of 1216 rows, nnz: 6000000
    Wrote a total of 1441 rows, nnz: 7000000
    
    ...
    

    NOTE: You may want also to checkout GraphLab's collaborative filtering library: here. GraphLab has a 100% compatible SVD solver to Mahout, with performance gains up to x50 times faster. I have created Java code to convert Mahout sequence files into Graphlab's format and back. Email me and I will send you the code.

    Tuesday, February 1, 2011

    Mahout on CMU OpenCloud

    This post explains how to run Mahout on top of CMU OpenCloud.

    1) log into the cloud login node
    ssh -L 8888:proxy.opencloud:8888 login.cloud.pdl.cmu.local.

    2) copy mahout directory tree into your home folder. 

    3) Run Mahout example
    cd mahout-0.4/
    export JAVA_HOME=/usr/lib/jvm/java-6-sun/
     ./examples/bin/build-reuters.sh

    You should see:
    sh -x ./examples/bin/build-reuters.sh
    11/02/01 15:13:27 INFO driver.MahoutDriver: Program took 225915 ms
    + ./bin/mahout seqdirectory -i ./examples/bin/work/reuters-out/ -o ./examples/bin/work/reuters-out-seqdir -c UTF-8 -chunk 5
    Running on hadoop, using HADOOP_HOME=/usr/local/sw/hadoop
    HADOOP_CONF_DIR=/etc/hadoop/conf/global
    11/02/01 15:13:38 INFO driver.MahoutDriver: Program took 10087 ms
    + ./bin/mahout seq2sparse -i ./examples/bin/work/reuters-out-seqdir/ -o ./examples/bin/work/reuters-out-seqdir-sparse
    Running on hadoop, using HADOOP_HOME=/usr/local/sw/hadoop
    HADOOP_CONF_DIR=/etc/hadoop/conf/global
    11/02/01 15:13:40 INFO vectorizer.SparseVectorsFromSequenceFiles: Maximum n-gram size is: 1
    11/02/01 15:13:40 INFO vectorizer.SparseVectorsFromSequenceFiles: Minimum LLR value: 1.0
    11/02/01 15:13:40 INFO vectorizer.SparseVectorsFromSequenceFiles: Number of reduce tasks: 1
    11/02/01 15:13:41 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
    11/02/01 15:13:42 INFO input.FileInputFormat: Total input paths to process : 3
    11/02/01 15:13:47 INFO mapred.JobClient: Running job: job_201101170028_1733
    11/02/01 15:13:48 INFO mapred.JobClient:  map 0% reduce 0%
    11/02/01 15:17:49 INFO mapred.JobClient:  map 33% reduce 0%
    11/02/01 15:17:55 INFO mapred.JobClient:  map 66% reduce 0%
    11/02/01 15:18:01 INFO mapred.JobClient:  map 100% reduce 0%
    11/02/01 15:18:08 INFO mapred.JobClient: Job complete: job_201101170028_1733
    11/02/01 15:18:08 INFO mapred.JobClient: Counters: 6
    11/02/01 15:18:08 INFO mapred.JobClient:   Job Counters
    11/02/01 15:18:08 INFO mapred.JobClient:     Rack-local map tasks=5
    11/02/01 15:18:08 INFO mapred.JobClient:     Launched map tasks=5
    11/02/01 15:18:08 INFO mapred.JobClient:   FileSystemCounters
    11/02/01 15:18:08 INFO mapred.JobClient:     HDFS_BYTES_READ=13537042
    11/02/01 15:18:08 INFO mapred.JobClient:     HDFS_BYTES_WRITTEN=11047110
    11/02/01 15:18:08 INFO mapred.JobClient:   Map-Reduce Framework
    11/02/01 15:18:08 INFO mapred.JobClient:     Map input records=16115
    11/02/01 15:18:08 INFO mapred.JobClient:     Spilled Records=0
    11/02/01 15:18:08 WARN mapred.JobClient: Use GenericOptionsParser for parsing the arguments. Applications should implement Tool for the same.
    11/02/01 15:18:09 INFO input.FileInputFormat: Total input paths to process : 3
    11/02/01 15:18:15 INFO mapred.JobClient: Running job: job_201101170028_1736
    11/02/01 15:18:16 INFO mapred.JobClient:  map 0% reduce 0%
    ...

    Tuesday, January 25, 2011

    Mahout on Amazon EC2 - part 2 - Running Hadoop on a single node

    Following part 1 of this posting which explained how to install Mahout and Hadoop on Amazon EC2.


    We start by testing logistic regression


    1) Launch Amazon AMI image you constructed using the explanation in part 1 of this post.

    2) Run Hadoop using
    # $HADOOP_HOME/bin/hadoop namenode -format
    # $HADOOP_HOME/bin/start-all.sh
    # jps     // you should see all 5 Hadoop processes (NameNode, SecondaryNameNode, DataNode, JobTracker, TaskTracker)
    

    3) Run logistic regression example

    cd /usr/local/mahout-0.4/
    ./bin/mahout org.apache.mahout.classifier.sgd.TrainLogistic --passes 100 --rate 50 --lambda 0.001 --input examples/src/main/resources/donut.csv --features 21 --output donut.model --target color --categories 2 --predictors x y xx xy yy a b c --types n n
    

    You should see the following output:

    11/01/25 14:42:45 WARN driver.MahoutDriver: No org.apache.mahout.classifier.sgd.TrainLogistic.props found on classpath, will use command-line arguments only
    21
    color ~ 0.353*Intercept Term + 5.450*x + -1.671*y + -4.740*xx + 0.353*xy + 0.353*yy + 5.450*a + 2.765*b + -24.161*c
          Intercept Term 0.35319
                       a 5.45000
                       b 2.76534
                       c -24.16091
                       x 5.45000
                      xx -4.73958
                      xy 0.35319
                       y -1.67092
                      yy 0.35319
    
        2.765337737     0.000000000    -1.670917299     0.000000000     0.000000000     0.000000000     5.449999190     0.000000000   -24.160908591    -4.739579336     0.353190637     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000     0.000000000
    
    11/01/25 14:42:46 INFO driver.MahoutDriver: Program took 1016 ms
    

    Now we run alternating matrix factorization. Based on instructions by Sebastian Schelter (see https://issues.apache.org/jira/browse/MAHOUT-542).
    A related GraphLab implementation is found here

    0) Download the patch MAHOUT-542.patch from the above webpage.
    Installl it using the command
    cd /usr/local/mahout-0.4/src/
    wget https://issues.apache.org/jira/secure/attachment/12469671/MAHOUT-542-5.patch
    patch -p0 < MAHOUT-542-5.patch
    
    1) Get the movie lens 1M movie dataset
    cd /usr/local/mahout-0.4/
    wget http://www.grouplens.org/system/files/million-ml-data.tar__0.gz
    tar xvzf million-ml-data.tar__0.gz
    
    2) Convert dataset to csv format
    cat ratings.dat |sed -e s/::/,/g| cut -d, -f1,2,3 > ratings.csv
    cd /usr/local/hadoop-0.20.2/
    ./bin/hadoop fs -copyFromLocal /path/to/ratings.csv ratings.csv
    ./bin/hadoop fs -ls
    
    
    Should see something like
    /user/ubuntu/ratings.csv
    
    
    
    3) # create a 90% percent training set and a 10% probe set
    /usr/local/mahout-0.4$ ./bin/mahout splitDataset  --input /user/ubuntu/ratings.csv --output /user/ubuntu/myout --trainingPercentage 0.9 --probePercentage 0.1
    
    The output should look like:
    Running on hadoop, using HADOOP_HOME=/usr/local/hadoop-0.20.2/
    HADOOP_CONF_DIR=/usr/local/hadoop-0.20.2/conf
    11/01/27 01:09:39 WARN driver.MahoutDriver: No splitDataset.props found on classpath, will use command-line arguments only
    11/01/27 01:09:39 INFO common.AbstractJob: Command line arguments: {--endPhase=2147483647, --input=/user/ubuntu/ratings.csv, --output=/user/ubuntu/myout, --probePercentage=0.1, --startPhase=0, --tempDir=temp, --trainingPercentage=0.9}
    11/01/27 01:09:40 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
    11/01/27 01:09:40 INFO input.FileInputFormat: Total input paths to process : 1
    11/01/27 01:09:40 INFO mapred.JobClient: Running job: job_local_0001
    11/01/27 01:09:40 INFO input.FileInputFormat: Total input paths to process : 1
    11/01/27 01:09:41 INFO mapred.MapTask: io.sort.mb = 100
    11/01/27 01:09:41 INFO mapred.MapTask: data buffer = 79691776/99614720
    11/01/27 01:09:41 INFO mapred.MapTask: record buffer = 262144/327680
    11/01/27 01:09:42 INFO mapred.JobClient:  map 0% reduce 0%
    11/01/27 01:09:42 INFO mapred.MapTask: Spilling map output: record full = true
    11/01/27 01:09:42 INFO mapred.MapTask: bufstart = 0; bufend = 5970616; bufvoid = 99614720
    11/01/27 01:09:42 INFO mapred.MapTask: kvstart = 0; kvend = 262144; length = 327680
    11/01/27 01:09:42 INFO util.NativeCodeLoader: Loaded the native-hadoop library
    
    4)# run distributed ALS-WR to factorize the rating matrix based on the training set
    bin/mahout parallelALS --input /user/ubuntu/myout/trainingSet/ --output /tmp/als/out --tempDir /tmp/als/tmp --numFeatures 20 --numIterations 10 --lambda 0.065
    ...
    11/01/27 02:40:28 INFO mapred.JobClient:     Spilled Records=7398
    11/01/27 02:40:28 INFO mapred.JobClient:     Map output bytes=691713
    11/01/27 02:40:28 INFO mapred.JobClient:     Combine input records=0
    11/01/27 02:40:28 INFO mapred.JobClient:     Map output records=3699
    11/01/27 02:40:28 INFO mapred.JobClient:     Reduce input records=3699
    11/01/27 02:40:28 INFO driver.MahoutDriver: Program took 1998612 ms
    
    5)# measure the error of the predictions against the probe set
    usr/local/mahout-0.4$ bin/mahout evaluateALS --probes /user/ubuntu/myout/probeSet/ --userFeatures /tmp/als/out/U/ --itemFeatures /tmp/als/out/M/
    Running on hadoop, using HADOOP_HOME=/usr/local/hadoop-0.20.2/
    HADOOP_CONF_DIR=/usr/local/hadoop-0.20.2/conf
    11/01/27 02:42:37 WARN driver.MahoutDriver: No evaluateALS.props found on classpath, will use command-line arguments only
    11/01/27 02:42:37 INFO common.AbstractJob: Command line arguments: {--endPhase=2147483647, --itemFeatures=/tmp/als/out/M/, --probes=/user/ubuntu/myout/probeSet/, --startPhase=0, --tempDir=temp, --userFeatures=/tmp/als/out/U/}
    
    ...
    
    Probe [99507], rating of user [4510] towards item [2560], [1.0] estimated [1.574626183998361]
    Probe [99508], rating of user [4682] towards item [171], [4.0] estimated [4.073943928686575]
    Probe [99509], rating of user [3333] towards item [1215], [5.0] estimated [4.098295242062813]
    Probe [99510], rating of user [4682] towards item [173], [2.0] estimated [1.9625234269143972]
    RMSE: 0.8546120366924382, MAE: 0.6798083002225481
    11/01/27 02:42:50 INFO driver.MahoutDriver: Program took 13127 ms
    
    Useful HDFS commands * View the current state of the file system
    ubuntu@domU-12-31-39-00-18-51:/usr/local/hadoop-0.20.2$ ./bin/hadoop dfsadmin -report
    Configured Capacity: 10568916992 (9.84 GB)
    Present Capacity: 3698495488 (3.44 GB)
    DFS Remaining: 40173568 (38.31 MB)
    DFS Used: 3658321920 (3.41 GB)
    DFS Used%: 98.91%
    Under replicated blocks: 56
    Blocks with corrupt replicas: 0
    Missing blocks: 0
    
    -------------------------------------------------
    Datanodes available: 1 (1 total, 0 dead)
    
    Name: 127.0.0.1:50010
    Decommission Status : Normal
    Configured Capacity: 10568916992 (9.84 GB)
    DFS Used: 3658321920 (3.41 GB)
    Non DFS Used: 6870421504 (6.4 GB)
    DFS Remaining: 40173568(38.31 MB)
    DFS Used%: 34.61%
    DFS Remaining%: 0.38%
    Last contact: Tue Feb 01 21:10:15 UTC 2011
    
    * Delete a directory
    ubuntu@domU-12-31-39-00-18-51:/usr/local/hadoop-0.20.2$ ./bin/hadoop fs -rmr temp/markedPreferences
    Deleted hdfs://localhost:9000/user/ubuntu/temp/markedPreferences
    

    Monday, January 24, 2011

    Mahout/Hadoop on Amazon EC2 - part 1 - Installation

    This post explains how to install Mahout ML framework on top of Amazon EC2 (Ubuntu based machine).
    The notes are based on older Mahout notes: https://cwiki.apache.org/MAHOUT/mahout-on-amazon-ec2.html which are unfortunately outdated

    The next of the post (part 2) explains how to run two Mahout applications:
    logistic regression and alternating least squares.

    Note: part 5 of this post, explains how to make the same installation on top of
    ec2 high computing node (CentOS/Redhat machine). Unfortunately, several steps
    are different..

    Part 6 of this post explains how to fine tune performance on large cluster.

    Full procedure should take around 2-3   hours.. :-(

    To confuse the users, Amazon has 5 types of IDs:
    - Your email and password for getting into the AWS console
    - Your AWS string name and private key string
    - Your public/private key pair
    - Your X.509 certificate (another private/public key pair)
    - Your Amazon ID (12 digit number) which is very hard to find on their website
    Make sure you have all your IDS ready, if you did not do it yet, generate the keys using AWS console.

    1) select and launch instance ami-08f40561 from Amazon AWS console. Alternatively you can select any other Ubuntu based 64 bit image.
    TIP: It is recommended using EBS backed image, since saving your work at the end will be made way easier.

    2) verify java is installed correctly - some libs are missing in the ami
    sudo apt-get install openjdk-6-jdk
    sudo apt-get install openjdk-6-jre-headless
    sudo apt-get install openjdk-6-jre-lib
    

    3) In the root home directory evaluate:
    # sudo apt-get update
    # sudo apt-get upgrade
    # sudo apt-get install python-setuptools
    # sudo easy_install "simplejson==2.0.9"
    # sudo easy_install "boto==1.8d"
    # sudo apt-get install ant
    # sudo apt-get install subversion
    # sudo apt-get install maven2
    

    4) for getting hadoop source
    # wget http://apache.cyberuse.com//hadoop/core/hadoop-0.20.2/hadoop-0.20.2.tar.gz 
    # tar vxzf hadoop-0.20.2.tar.gz
    # sudo  mv hadoop-0.20.2 /usr/local/
    

    A comment: I once managed to install 0.21.0, but after the EC2 node was killed and restarted
    Mahout refused to work any more. So I reverted to Hadoop 0.20.2

    add the following to $HADOOP_HOME/conf/hadoop-env.sh
    export JAVA_HOME=/usr/lib/jvm/java-6-openjdk/
    # The maximum amount of heap to use, in MB. Default is 1000
    export HADOOP_HEAPSIZE=2000
    

    add the following to $HADOOP_HOME/conf/core-site.xml and also $HADOOP_HOME/conf/mapred-site.xml
    <pre class="xml" name="code"><configuration>     
    <property>     
    <name>fs.default.name</name>     
    <value>hdfs://localhost:9000</value>   
    </property>   <property>     
    <name>mapred.job.tracker</name>      
    <value>localhost:9001</value>    
    </property>  
     <property>      
    <name>dfs.replication</name>      
     <value>1</value>           
      </property>   
    <property> 
     <name>hadoop.tmp.dir</name> 
    <value>/mnt/tmp/</value>  
    </property>  
    </configuration></pre>
      
    
    Edit the file hdfs-site.xml
    <pre class="xml" name="code"><configuration>
     <property>  
      <name>hadoop.tmp.dir</name> 
      <value>/mnt/tmp/</value>   
     </property>   
    <property>   
     <name>dfs.data.dir</name>
     <value>/mnt/tmp2/</value>
    </property>  
    <property> 
     <name>dfs.name.dir</name>
     <value>/mnt/tmp3/</value> 
    </property> 
    </configuration> 
    </pre>
     
    

    Note: pointing the directories to /mnt is done since on Amazon EC2 regular instances has 400GB free space there (vs. only 10GB free space on root partition). You may
    need to change permissions of /mnt in so this file system will be writable by Hadoop.
    So execute the following command:
    sudo chmod 777 /mnt
    


    Set up authorized keys for localhost login w/o passwords and format your name node
    # ssh-keygen -t dsa -P '' -f ~/.ssh/id_dsa
    # cat ~/.ssh/id_dsa.pub >> ~/.ssh/authorized_keys
    


    5)Add the following to your .profile
    export JAVA_HOME=/usr/lib/jvm/java-6-openjdk
    export HADOOP_HOME=/usr/local/hadoop-0.20.2
    export HADOOP_CONF_DIR=/usr/local/hadoop-0.20.2/conf
    export MAHOUT_HOME=/usr/local/mahout-0.4/
    export MAHOUT_VERSION=0.4-SNAPSHOT
    export MAVEN_OPTS=-Xmx1024m
    





  • 6) Checkout and build Mahout from trunk. ify that the paths on .profile point to the exact version you downloaded

    svn co http://svn.apache.org/repos/asf/mahout/trunk mahout
    cd mahout
    mvn clean install
    cd ..
    sudo mv mahout /usr/local/mahout-0.4
    

    Note: I am getting a lot of questions about the mvn compilation.
    a) On windows based machines, it seems that running a Linux VM makes some
    of the tests fail. Try to compile with the flag -DskipTests
    b) If compilation fails, you can try and download compiled jars from
    http://mirror.its.uidaho.edu/pub/apache//mahout/0.4/ (the compiled jar are
    in the files without "src" in the filename). Just open the tgz and place it
    on /usr/local/mahout-0.4/ instead of the compilation step above.


    7) Install other required stuff (optional: in the Amazon EC2 image I am using
    those libraries are preinstalled).
    sudo apt-get install wget alien ruby libopenssl-ruby1.8 rsync curl
    

    8) Run Hadoop, just to prove you can, and test Mahout by building the Reuters dataset on it. Finally, delete the files and shut it down.

    $HADOOP_HOME/bin/hadoop namenode -format
    $HADOOP_HOME/bin/start-all.sh
    jps     // you should see all 5 Hadoop processes (NameNode, SecondaryNameNode, DataNode, JobTracker, TaskTracker)
    cd $MAHOUT_HOME
    ./examples/bin/build-reuters.sh
    $HADOOP_HOME/bin/stop-all.sh
    rm -rf /tmp/*   // delete the Hadoop files







  • Remove the single-host stuff you added to $HADOOP_HOME/conf/core-site.xml and $HADOOP_HOME/conf/mapred-site.xml in step #6b and verify you are happy with the other conf file settings. The Hadoop startup scripts will not make any changes to them. In particular, upping the Java heap size is required for many of the Mahout jobs.
    // edit $HADOOP_HOME/conf/mapred-site.xml to include the following:
    <property>
       <name>mapred.child.java.opts</name>
       <value>-Xmx2000m</value>
    </property>


    9) Allow for Hadoop to run even if you will work on a different EC2 machine:
    echo "NoHostAuthenticationForLocalhost yes" >>~/.ssh/config
    


    If everything went well, you may want to bundle the output into an AMI image, so next time you will not need to install everything from scratch:
    10) Install Amazon AMI tools
    a) Edit the file /etc/apt/sources.list
    and uncomment all the lines with multiverse (note: you need to call the editor as root!)
    b) update the repositories
    sudo apt-get update
    c) Install ami and api tools
    sudo apt-get install ec2-ami-tools ec2-api-tools
    
    Thanks Kevin for this fix!

    11) In order to save your work, you need to bundle and save the image.
    Here there are two alternatives. If you started EBS backed image, you can simply use the Amazon AWS user interface, right mouse click on the running instance and select "save instance".
    If the image is not EBS, you will need to do it manually:

    - note you need to use the private key of the x.509 certificate and not the private key of the public private key pair!!!!!!!

    [All the following commands should span one shell line..]

    First you need to create a bucket named mahoutbucket using the Amazon AWS console
    under S3 tab.

    sudo ec2-bundle-vol -k /mnt/pk-<your private X.509 key>.pem -c /mnt/cert-<your public x.509 key>.pem -u <Your AWS ID (12 digit number)> -d /mnt -p mahout
    sudo ec2-upload-bundle -b mahoutbucket -m /mnt/mahout.manifest.xml -a <Your AWS String> -s <Your AWS string password> 
    sudo ec2-register -K /mnt/pk-<Your X.509 private key>.pem -C /mnt/cert-<Your X.509 public certificate>.pem --name mahoutbucket/  mahoutbucket/mahout.manifest.xml
    
    If you are lucky -You will get a result of the type: IMAGE   ami-XXXXXXX
    where XXXXXXX is the generated image number.

    More detailed explanations about this procedure, along with many potential pitfalls are found
    in my blog post here.
    Thanks to Kevin and Selwyn!