Saturday, 6 April 2013

An R implementation of the Weighted Kernel K-Means

The only R implementation of the kernel k-means I found online is that of  kernlab package. According to the documentation of the method kkmeans, the method accepts a parameter called centers that is described as:
"Either the number of clusters or a matrix of initial cluster centers. If the first a random initial partitioning is used."
It was not clear to me what kind of centers I should pass to the method If I wanted to use my own initial cluster assignments. Passing centers in the original feature space does not make any sense since those centers are not necessarily good centers in the kernel space. According to my understanding, the proper way for passing initial clustering to kernel k-means is to specify the cluster Id of each data sample. The description above of the parameter centers was confusing to me. Moreover, the method returns the final centers of the clusters. The returned centers are of the same dimension as the original feature space which, to my knowledge, cannot be obtained unless an approximate (or an expensive full) decomposition is used. I posted my inquiry to R mailing list but, I did not get any useful answer.

Another limitation with the kkmeans method is that it does not allow specifying a convergence criteria (e.g. maximum number of iterations). Controlling both the initial centers and the convergence criteria is important indeed for comparing different variations. Moreover, I was getting error messages that I could not find any clue how to fix them. An example on such error messages is:

 "Error in crossprod(x[vgr[[i]], vgr[[i]], drop = FALSE], w[vgr[[i]]]) :  
  object 'vgr' not found"  

So, I had to write my own R implementation of the weighted kernel k-means.The code is available here.

In my implementation, I am handling the empty cluster problem by randomly picking a sample that was assigned to the empty cluster in the previous iteration and moving it back to that cluster. It might not be the optimal solution, but it is the cheapest in terms of memory and time requirements. The implementation trades off time for space. That is why I am not using the triangular inequality described in [1].


[1] Dhillon et. al. "Kernel k-means, Spectral Clustering and Normalized Cuts". In KDD'04.

Wednesday, 3 October 2012

Running a Jar File on Amazon Elastic MapReduce: A Step by Step Guide

In this post I am describing the steps of running hadoop jar files on Amazon Elastic MapReduce (EMR).

1- Install EMR Command Line Tool as described here.

2- Upload the dataset file (dataset.txt) and the jar file (myjar.jar) to Amazon S3.
     https://console.aws.amazon.com/s3/home

3-  cd to the installation directory of the command line tool

4- Launch the EMR Cluster

./elastic-mapreduce --create --alive --name "cluster name" --hadoop-version 1.0.3 --num-instances 10 --instance-type m1.large --bootstrap-action s3://elasticmapreduce/bootstrap-actions/configure-hadoop --args "-m","mapred.reduce.child.java.opts=-Xmx2g"

The command above starts an EMR cluster using hadoop version 1.0.3. The cluster consists of 10 nodes (1 master and 9 slaves). The 10 nodes are all m1.large EC2 instances. I am also using a bootstrap action to enlarge the JVM heap size of the reducer (more details: in a previous post here).
The job Id (e.g. j-4CYDKHIUAQL7) will be printed out after running the command above. 

5- Wait till the cluster is started and bootstrapped. You can use the command

./elastic-mapreduce --list j-4CYDKHIUAQL7

to pull the state of the cluster. An output like this:

j-4CYDKHIUAQL7      STARTING                                                       cluster_name

means that the cluster is still starting. Wait till STARTING is replaced with WAITING.

6- Use the following command to get the public DNS address of the master node

./elastic-mapreduce --describe j-4CYDKHIUAQL7 | grep MasterPublicDnsName | cut -d'"' -f4

7- Log in to the master node to copy the dataset from S3 to HDFS installed on the cluster. You might decide to run the Jar on the dataset on S3. That is perfectly fine if a single iteration is carried out over the dataset. But, when multiple iterations over the dataset are expected, performance-wise its better to copy the dataset to HDFS only once before starting the job.

ssh -A hadoop@ec2-xx-xx-xxx-xx.compute-1.amazonaws.com

8- Set the configurations of s3cmd (a tool to 'get' & 'put' files from/to S3). s3cmd will be installed on the master EMR node. You just need to set its configurations

hadoop@ip-xx-xx-xxx-xxx:~$ s3cmd --configure

The two required configurations are the access key and the secret key of your AWS account. The other configurations can be set to their default values.

9- Copy the dataset from S3 to the local storage of the master node

hadoop@ip-xx-xx-xxx-xxx:~$ s3cmd get s3://mys3bucket/dataset.txt dataset.txt

10- Copy the dataset from the local storage to HDFS.

hadoop@ip-xx-xx-xxx-xxx:~$ hadoop dfs -copyFromLocal dataset.txt /dataset.txt

11- Now you can log out of the master node. Back to the command line tool of EMR and run the jar file:

./elastic-mapreduce -j j-4CYDKHIUAQL7 --jar s3n://mys3bucket/myjar.jar --args "/dataset.txt"

Here I am passed the path of the dataset on HDFS as the first argument to the jar file.

12- You can monitor the job execution via EMR web console or even using the command in step 5.

That is all about it!

Friday, 14 September 2012

Fixing "Inconsistent split cardinality" problem in Mahout Matrix Multiplication

I was trying out mahout's matrix multiplication facility that  can be invoked via the times() method of a DistributedRowMatrix instance. i.e.

1:  final Configuration conf = new Configuration();  
2:  DistributedRowMatrix A = new DistributedRowMatrix(   
3:                    new Path("/A"), new Path("/tmp/A"), 100, 60000);  
4:  A.setConf(conf);  
5:  DistributedRowMatrix B = new DistributedRowMatrix(   
6:                    new Path("/B"), new Path("/tmp/B"), 100, 30);  
7:  B.setConf(conf);  
8:  DistributedRowMatrix C = A.times(B);  // C = A'*B

I got the following exception:

12/09/01 23:33:46 ERROR security.UserGroupInformation: PriviledgedActionException as:ahmed cause:java.io.IOException: Inconsistent split cardinality from child 1 (2/1)
Exception in thread "main" java.io.IOException: Inconsistent split cardinality from child 1 (2/1)
    at org.apache.hadoop.mapred.join.Parser$CNode.getSplits(Parser.java:369)
    at org.apache.hadoop.mapred.join.CompositeInputFormat.getSplits(CompositeInputFormat.java:117)
    at org.apache.hadoop.mapred.JobClient.writeOldSplits(JobClient.java:989)
    at org.apache.hadoop.mapred.JobClient.writeSplits(JobClient.java:981)
    at org.apache.hadoop.mapred.JobClient.access$600(JobClient.java:174)
    at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:897)
    at org.apache.hadoop.mapred.JobClient$2.run(JobClient.java:850)
    at java.security.AccessController.doPrivileged(Native Method)
    at javax.security.auth.Subject.doAs(Subject.java:396)
    at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1121)
    at org.apache.hadoop.mapred.JobClient.submitJobInternal(JobClient.java:850)
    at org.apache.hadoop.mapred.JobClient.submitJob(JobClient.java:824)
    at org.apache.hadoop.mapred.JobClient.runJob(JobClient.java:1261)
    at org.apache.mahout.math.hadoop.DistributedRowMatrix.times(DistributedRowMatrix.java:189)
    at baselines.ReconstructionError.clacReconstructionErr(ReconstructionError.java:45)
    at baselines.Main.runRandomSelection(Main.java:108)
    at baselines.Main.main(Main.java:154)
    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.RunJar.main(RunJar.java:156)


Checking out the source of the MatrixMultiplicationJob, I found that the job's input format is  CompositeInputFormat of the matrices being multiplied. The CompositeInputFormat joins the input records of the two matrices on the Row Number and provides the mapper with an IntWritable (the row number i) and a TubleWritable (combined record of row #i of the first matrix and row #i of the second matrix). In order for the CompositeInputFormat to be able to join the two matrices, the two matrices should be sorted and partitioned the same. And that was the cause of the exception above. The two matrices were not partitioned the same. The first matrix was written to a single file (original format) while the second matrix was the output of a previous MapReduce job with number of reducers > 1.

Although It is not an uncommon scenario, the documentation of the times() method mentions nothing about that "limitation". My solution was to repartition one of the matrices using the same partitioner (default hashing function of Hadoop's Partitioner + number of reducers) of the other. In my case, one of the matrices (A) was pretty much larger than the other one (B). So to minimize the partitioning overhead, I decided to repartition the small one (B). One way to know about the number of A's partitions is to count the number of the files under A's directory (each reducer generates a single file).

The repartitioning method is simple. we just use the default Mapper and the default Reducer and set the number of the reducers to the number of A's partitions. Here is the my implementation:
1:  public static void repartitionMatrix(Path input, Path output, int numPartitions) throws Exception {  
2:    final Configuration conf = new Configuration();  
3:    Job job = new Job(conf);  
4:    job.setJarByClass(Helpers.class);  
5:    FileInputFormat.addInputPaths(job, input.toString());  
6:    FileOutputFormat.setOutputPath(job, output);  
7:    job.setMapperClass(Mapper.class);  
8:    job.setReducerClass(Reducer.class);  
9:    job.setOutputKeyClass(IntWritable.class);  
10:    job.setOutputValueClass(VectorWritable.class);  
11:    job.setInputFormatClass(SequenceFileInputFormat.class);  
12:    job.setOutputFormatClass(SequenceFileOutputFormat.class);  
13:    job.setNumReduceTasks(numPartitions);  
14:    job.waitForCompletion(false);  
15:    if (!job.isSuccessful())  
16:        throw new RuntimeException("Repartition Job is unsuccessful");  
17:  }  


Configuring the Reducer Heap Size on Elastic MapReduce

I was running mahout MatrixMultiplicationJob on Amazon Elastic MapReduce (EMR) to multiply two matrices of sizes (150k x 8.2m) and (100 x 150k  ). My cluster consisted of 1 master node and 19 core nodes. All of the 20 nodes were m1.large ec2 instances each o which had 7.5GB of memory according to EC2 Docs. Despite the fairly suitable memory size (as I thought), I got an out of memory error in the reduce task.


Task attempt_201209140440_0004_m_000024_0 failed to report status for 601 seconds. Killing!
Error: java.lang.OutOfMemoryError: Java heap space
 at org.apache.mahout.math.map.OpenIntDoubleHashMap.rehash(OpenIntDoubleHashMap.java:434)
 at org.apache.mahout.math.map.OpenIntDoubleHashMap.put(OpenIntDoubleHashMap.java:387)
 at org.apache.mahout.math.RandomAccessSparseVector.setQuick(RandomAccessSparseVector.java:139)
 at org.apache.mahout.math.AbstractVector.assign(AbstractVector.java:560)
 at org.apache.mahout.math.hadoop.MatrixMultiplicationJob$MatrixMultiplicationReducer.reduce(MatrixMultiplicationJob.java:161)
 at org.apache.mahout.math.hadoop.MatrixMultiplicationJob$MatrixMultiplicationReducer.reduce(MatrixMultiplicationJob.java:147)
 at org.apache.hadoop.mapred.Task$OldCombinerRunner.combine(Task.java:1436)
 at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.doInMemMerge(ReduceTask.java:2815)
 at org.apache.hadoop.mapred.ReduceTask$ReduceCopier$InMemFSMergeThread.run(ReduceTask.java:2753)

Then, I tried to increase the JVM Heap size of the reduce tasks. EMR allows users to use a set of Bootstrap Actions to overwrite the configurations and the environment argument of EMR. Bootstrap Actions can be specified in the --create command (more about EMR commands here). The format of a bootstrap action should be like this:

 --bootstrap-action "s3://myawsbucket/FileName" --args "arg1","arg2"   

FileName refers to a script file that is executed on the cluster to do whatever changes. The required changed should be specified via the arguments of that script.

To alter any of Hadoop configurations, the script file that should be used is:
s3://elasticmapreduce/bootstrap-actions/configure-hadoop 

and the arguments should be formated as "-configFile","parameter=newValue". configFile determines Hadoop configurations file that contains the default value of the parameter being altered. Here are Hadoop configuration files and the corresponding configFile:

File Name configFile
hadoop-site.xml s
hadoop-default.xml d
core-site.xml c
hdfs-site.xml h
mapred-site.xml m

I wanted to increase the heap size of the reducer JVM. So, I had to alter the configuration parameter mapred.reduce.child.java.opts to set the maximum heap size to a larger value (-Xmx2g). That parameter is one of the configurations in mapred-site.xml. So, by bootstrap command was:

 --bootstrap-action s3://elasticmapreduce/bootstrap-actions/configure-hadoop --args "-m","mapred.reduce.child.java.opts=-Xmx2g"

The full --create command was:
 ./elastic-mapreduce --create --alive --name "GCSS Eval-mem" --hadoop-version 1.0.3 --num-instances 20 --instance-type m1.large --bootstrap-action s3://elasticmapreduce/bootstrap-actions/configure-hadoop --args "-m","mapred.reduce.child.java.opts=-Xmx2g"