KDD Cup progress update - We are in the 5th place!

I am proud to report that yesterday JustinYan, a graduate student from the Institute of Automation, Chinese Academy of Sciences, a member of the LeBuSiShu team, suggested that I will join their team.
The reason is that LeBuSiShu is using GraphLab ALS solution as one of the components in their final result. Currently LeBuSiShu is the 5th group in the leaderboard (out of 100 groups!) with RMSE = 22.2820.

Anyway, of course I was honored to accept as well as quite excited! Although when writing the GraphLab collaborative filtering code of matrix factorization I had no competition in mind, it is always nice to know that someone finds your solution useful.

Besides of JustinYun, another nice team member in the LeBuSiShu group is: Yao Wu, National Laboratory of Pattern Recognition, Institute of Automation, Chinese Academic of Sciences.

A quick update 6/23: We are actually now at the 4th place! The Chinese group is an amazing team, they simply work around the clock. I myself am running some fancy MCMC methods for slightly improving prediction, but the Chinese team is responsible of 99.9% of the progress.

Yahoo! KDD Cup using Graphlab - Track 2?

I was delighted to hear from Suhrid Balakrishnan from AT&T Labs, that he is using GraphLab pmf for factorizing a linear model for Yahoo! KDD cup - track 2. Initially I focused only on track1, but it seems that Graphlab is potentially useful for track2 as well.

Overall, this month I am aware of 20 installations of Graphlab pmf code on various research groups all over the world. Specifically I got feedback from University of Austin, University of the Aegean, University of Macedonia, Carnegie Mellon University and AT&T Labs.

I got several valuable inputs regarding his experience with GraphLab I wanted to share and ask if anyone had the same experience.

1) It is recommended to download latest version from mercurial repository. See explanation:
in my previous post http://bickson.blogspot.com/2011/04/yahoo-kdd-cup-using-graphlab.html
I am constantly improving the matrix factorization code and adapting it to the KDD dataset. 

2) It is better to install itpp first, since Graphlab auto detects it and it saves much later trouble.

3) Suhrid got an interesting error when saving the factorized matrices U,V. It seems that on his Ubuntu system, factor ordering was somehow reversed.
The following matlab code solved this issue:

Ud=reshape(U(:),size(U,2),size(U,1));
Ud=Ud';

I have opened a google group for discussions and questions concerning KDD usage of GraphLab. Everyone is welcome to join:
* Group name: GraphLab KDD
* Group home page: http://groups.google.com/group/graphlab-kdd
* Group email address graphlab-kdd@googlegroups.com

GraphLab on BlackLight!!

I am super excited to report that GraphLab is up and running on BlackLight, the largest
shared memory computer in the world! With 32TB shared memory and 4,096 cores.

A tutorial for BlackLight is found here

I will soon post some performance results for matrix factorization algorithms, as we make more progress in testing.

Below you can find some instructions on how to install Graphlab on BlackLight, to those of you who are lucky enough to get an account.. :-)

GraphLab Installation

1) Login using ssh into  tg-login1.blacklight.psc.teragrid.org

2) Follow the instructions on http://graphlab.org/download.html to obtain
GraphLab code/

3)

module load cmake boost kyotocabinet IT++
cd graphlabapi
./configure --bootstrap --itpp_include_dir=${ITPP_INC} --itpp_static_link_dir=${ITPP_LIB} -D MKL_PATH=${MKL_PATH}
cd release
make -j 8

Note: Thanks to Joel Welling from Pittsburgh Supercomputing Center, who significantly helped simplifying installation as well as improving performance.

Example GraphLab PMF job
Create a file named kddcup.job with the following content:

#!/bin/csh
#PBS -l ncpus=16
#ncpus must be a multiple of 16
#PBS -l walltime=4:00:00                  
#PBS -j oe
#PBS -q batch
#PBS -m bea
set echo

ja

#move to my $SCRATCH directory
cd $SCRATCH

#copy executable to $SCRATCH
cp $HOME/graphlabapi/release/demoapps/pmf/pmf .

#run my executable
omplace -nt $PBS_NCPUS ./pmf kddcup 0 --scheduler="round_robin(max_iterations=20)" --float=true --zero=true --lambda=1 --D=150 --ncpus=$PBS_NCPUS --aggregatevalidation=true
cp $SCRATCH/kddcupt.kdd.out $HOME/$PBS_JOBID.kdd.out

ja -chlst

Submit this job using the command

qsub kddcup.job

Check the status of the job using the command

qstat 

Check remaining qouta:

bickson@tg-login1:~> xbanner


PSC Grantnumber: DMS110004P Teragrid Grantnumber: DMS110015
P.I. Name: Carlos Guestrin
 Resource        = BLACKLIGHT
 Charge ID       = ms3bdkp
 Start Date      = 02/10/2011
 Expiration Date = 02/10/2012
 Allocation      = 50000.00
 Remaining       = 48541.23
 Last Job        = 06/16/2011

Last Accounting Update: 06/16/2011

Yahoo! KDD Cup using Graphlab

I got the following question from an avid reader of this blog:

I was wondering whether you could give me some directions on how to
setup GraphLab to run on KDDCUP, especially with regards to the data format of the input files for training, validation and testing (i.e. creation of predictions for submission).

Many thanks,
Nicholas

==========================================

Nicholas Ampazis
Assistant Professor
Director, Intelligent Data Exploration and Analysis Laboratory (IDEAL)
Department of Financial and Management Engineering,
University of the Aegean
41 Koudouriotou street, Chios, 82100, Greece

I think this may interest some other people so I am posting the answer here. Currently GraphLab was tested with matrix factorization, but soon I will handle also tensor factorization (divides ranking of different times to groups) and also Monte Carlo Sampling on top of it. So there is a wide range of algorithms you can actually try out after you install GraphLab.

Installation

The best way to start is to download the code from mercurial repository - there is the latest version of the matrix factorization code. Source is found here: http://graphlab.org/download.html

Note that for the matrix factorization you will also need to install itpp (which relies on BLAS/LaPaCK).
Installation instructions for Linux 32 bit are here:http://bickson.blogspot.com/2011/06/graphlab-pmf-on-32-bit-linux.html and for Linux 64 bit are here: http://bickson.blogspot.com/2011/02/installing-blaslapackitpp-on-amaon-ec2.html


It is always better to install itpp first, before executing the ./configure script, that way GraphLab will automatically detect itpp and installation becomes simpler.


After downloading GraphLab you should configure using;

./configure --bootstrap

This should install cmake and boost if they are missing on your system.



Once you compile successfully, it means that the application code of the matrix factorization code is compiled as well, it will be found in the directory demoapps/pmf

Setting up the input files - method 1 - using Matlab
1) Download the file save_c_gl4a.m to your local directory using:
wget http://www.graphlab.ml.cmu.edu/save_c_gl4a.m
2) Download the KDD Yahoo! Cup files (track1 dataset) from: 
http://kddcup.yahoo.com/datasets.php
3) Use the following Matlab script to convert the text dataset in binary graphlab format:
(It may take a couple of hours to finish depends on your machine..)
Note that you need to run the script 3 times - for runmode=1 (training data)
runmode= 2 (validation data), runmode=3 (test data).

%Script for converting KDD CUP 2011 data, written by Danny Bickson, CMU
%Can be round in matlab or octave
nUsers=1000990;
nItems=624961;
nRatings=262810175;
nTrainRatings=252800275;
nProbeRatings=4003960;
nTestRatings=6005940;


runmode=3;

filname='';
outfile='';
ratings=0;
switch runmode
    case 1
        disp('converting kdd cup 2011 training data - track 1');
        filename='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/trainIdx1.txt';
        ratings=nTrainRatings;
        outfile='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/kddcup';
    case 2
        disp('converting kdd cup 2011 validation data - track 1');
        filename='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/validationIdx1.txt';
        ratings=nProbeRatings;
        outfile='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/kddcupe';
    case 3
        disp('converting kdd cup 2011 test data - track 1');
        filename='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/testIdx1.txt';
        ratings=nTestRatings;
        outfile='/mnt/bigbrofs/usr7/bickson/kddcup/track1/track1/kddcupt';
end

ff=fopen(filename,'r');
if (ff < 0)
 error('failed to open input file for reading');
end
fout = fopen(outfile,'w');
if (fout < 0)
 error('failed to open file for writing');
end
%write output file matrix market format header
fprintf(fout, '%%%%MatrixMarket matrix coordinate real general\n');
fprintf(fout,'%d %d %d\n', nUsers, nItems, ratings);
cnt=1;
for j=1:nUsers
    [a,num]=fscanf(ff,'%d|%d',2);
    assert(num==2);

    user=a(1);
    if (mod(j,1000)==0)
        disp(['user: ', num2str(user),' ratings: ', num2str(a(2))]);
    end
    if (runmode==3)
        assert(a(2)==6);
    end


    for i=1:a(2)
        b=-100;
        if (runmode<=2)
            [b,num]=fscanf(ff,'%d %d %d %d:%d:%d',6);
            assert(num==6);
        else
            [b,num]=fscanf(ff,'%d %d %d:%d:%d',5);
            assert(num==5);
        end

        if (runmode<=2)
            fprintf(fout, '%d %d %d %d\n', user+1, b(1)+1, b(2), b(3));
        else
            fprintf(fout, '%d %d %d %d\n', user+1, b(1)+1, 1, b(2));
        end
        cnt=cnt+1;
    end
end

assert(cnt==ratings+1);

fclose(fout);

Setting up the input files - method 2 - Python
http://bickson.blogspot.com/2011/04/yahoo-kdd-cup-using-graphlab-part-2.html

Running GraphLab

1. cd into graphlabapi/release/demoapps/pmf or graphlabapi/debug/demoapss/pmf 
(depends if you want to debug or not).

2. Link the generate files from preparing input file, named kddcup (training), kddcupe (validation) and kddcupt(test) into your working directory using

ln -s /path/to/track1/kddcup* .

3. Run GraphLab example:

<73|0>bickson@bigbro6:~/newgraphlab/graphlabapi/release/demoapps/pmf$ ./pmf kddcup 0 --ncpus=8 --float=true --zero=true --lambda=1 --D=20 --scheduler="round_robin(max_iterations=15)"
Setting run mode ALS_MATRIX
INFO   :pmf.cpp(main:1233): ALS_MATRIX starting

loading data file kddcup
Loading kddcup TRAINING
Matrix size is: 1000990 624961 1
Creating 252800275 edges...
...............................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................................
.....loading data file kddcupe
Loading kddcupe VALIDATION
Matrix size is: 1000990 624961 1
Creating 4003960 edges...
.....................loading data file kddcupt
Loading kddcupt TEST
Matrix size is: 1000990 624961 6649
Creating 6005940 edges...
...............................setting regularization weight to 1
PTF_ALS for matrix (1000990, 624961, 6649):252800275.  D=20
pU=1, pV=1, 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=4.85456e+11, TRAIN RMSE=61.9728 TEST RMSE=75.7258.
 Entering last iter with 4
442.833) Iter ALS 4  Obj=5.91422e+10, TRAIN RMSE=21.6079 TEST RMSE=22.7737.
Entering last iter with 5
546.91) Iter ALS 5  Obj=5.71611e+10, TRAIN RMSE=21.2429 TEST RMSE=22.6440.
Entering last iter with 6
652.415) Iter ALS 6  Obj=5.63004e+10, TRAIN RMSE=21.0826 TEST RMSE=22.5745.
Entering last iter with 7
758.478) Iter ALS 7  Obj=5.58187e+10, TRAIN RMSE=20.9926 TEST RMSE=22.5299.
...

4. Explanation of basic runtime flags

    kddcup // input file name. Program optionally search also for optional inputs like validation and test data.
                  The convention is that validation data has the same file name ending with e (kddcupe) and test                   data ending with t (kddcupt).
    0 // the run mode. 0 stands for alternating least squares.
    --ncpus=XX // number of CPU used (should be equal to the number of cores you have)
    --lambda=0.1 //regularization weight for alternating least squares (this prameter should be fine tuned based on the problem
    --float=true //mandatory flag, indicating dataset is written in float format (yes, there is also an option for saving the dataset in double format if increased accuracy is desired)
    --zero=true //for KDDcup, this is mandatory, since some of the matrix/tensor values are zero. Without it the program will assert when there is zero matrix value. (In Netflix dataset there are no zero values).
    --scheduler="round_robin(max_iterations=XX)" the number of iterations to run
  --D=XX  // the width of the factorized matrix. As D is larger we get a better
approximation but slower running time.

5. More fancy runtime flags:

Other runmodes:
1 //Bayesian matrix factorization
2 //Bayesian tensor factorization
3 //Bayesian tensor factorization, supports for multiple ratings in different times.
4 //Alternating tensor factorization
--loadfactors=true // start initial guess from factors saved in previous run. Factor file name will be kddcup20.out where D=20 etc.
--scaling=100 //group the 6500 time units into groups of 100 (for tensor)
--truncating=2261 // remove unused time slots of ratings (for tensor)

Reading the Output
 When GraphLab detects a file name of kddcup, the output will be written to the file kddcupt.kdd.out
in the same working directory. This file name has the right format to be submitted into the contest website.

Additional output file of the name kddcupXX.out will be generated, where X is the width D of the approximating matrix. Instruction on how to read output files in Matlab are found on http://www.graphlab.ml.cmu.edu/pmf.html

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/

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.