Saturday, September 01, 2012

Learning Mahout : Clustering

The next section in the MIA book is Clustering. As with Recommenders, Mahout provides both in-memory and map-reduce versions of various clustering algorithms. However, unlike Recommenders, there are quite a few toolkits (like Weka or Mallet for example) which are more comprehensive than Mahout for small or medium sized datasets, so I decided to concentrate on the M/R implementations.

The full list of clustering algorithms available in Mahout at the moment can be found on its Wiki Page under the Clustering section. The ones covered in the book are K-Means, Canopy, Fuzzy K-Means, LDA and Dirichlet. All these algorithms expect data in the form of vectors, so the first step is to convert the input data into this format, a process known as vectorization. Essentially, clustering is the process of finding nearby points in n-dimensional space, where each vector represents a point in this space, and each element of a vector represents a dimension in this space.

It is important to choose the right vector format for the clustering algorithm. For example, one should use the SequentialAccessSparseVector for KMeans, sinc there is lot of sequential access in the algorithm. Other possibilities are the DenseVector and the RandomAccessSparseVector formats. The input to a clustering algorithm is a SequenceFile containing key-value pairs of {IntWritable, VectorWritable} objects. Since the implementations are given, Mahout users would spend most of their time vectorizing the input (and thinking about what feature vectors to use, of course).

Once vectorized, one can invoke the appropriate algorithm either by calling the appropriate bin/mahout subcommand from the command line, or through a program by calling the appropriate Driver's run method. All the algorithms require the initial centroids to be provided, and the algorithm iteratively perturbes the centroids until they converge. One can either guess randomly or use the Canopy clusterer to generate the initial centroids.

Finally, the output of the clustering algorithm can be read using the Mahout cluster dumper subcommand. To check the quality, take a look at the top terms in each cluster to see how "believable" they are. Another way to measure the quality of clusters is to measure the intercluster and intracluster distances. A lower spread of intercluster and intracluster distances generally imply "good" clusters. Here is code to calculate inter-cluster distance based on code from the MIA book.

There are various ways of improving clustering quality. The first, of course, is knowing your data well enough so you can choose good features to cluster on. For clustering text or real world (dirty) data, vector generation can be improved by removing noise and using a good weighting technique. Mahout allows you to specify your custom Lucene analyzers to its clustering subcommands for this. Cluster quality is also dependent on the measure used to calculate similarity between two feature vectors. Once again, Mahout supplies a large number of Distance Measure implementations (Chebyshev, Cosine, Mahalanobis, Manhattan, Minkowski, SquaredEuclidean, Euclidean, Tanimoto, Weighted Euclidean and Weighted Manhattan) and also allows you to specify your own if these don't suit your purposes. Within each dimension, points can be normalized to remove the effect of outliers - the normalization p-norm should match the p-norm used by the distance measure. Finally, if the dimensions are not comparable, such as number of bedrooms and price in dollars for a house, then one should normalize across dimensions, a process known as weighting (this should be done during the vectorization process, which you control fully).

Case Study - Reuters

The dataset used for this is the Reuters-21578 collection. I basically followed along with the book, trying out commands and making them work against a pseudo-distributed Hadoop installation on my notebook.

Vectors can be created using the following Mahout subcommands.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # create sequence file from directory of input files
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout seqdirectory \
  -c UTF-8 \
  -i reuters \          # input directory
  -o reuters-seqfiles   # output directory
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # create vectors from sequence file (no normalization)
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout seq2sparse \
  -i reuters-seqfiles \ # input: sequence files directory
  -o reuters-vectors \  # output: vectors
  -ow
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # this creates 7 sub directories under output
hduser@cyclone:mahout$ # df-count, dictionary.file*, frequency.file-*,
hduser@cyclone:mahout$ # tf-vectors, and tfidf-vectors
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # create vectors from sequence file (with normalization)
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout seq2sparse \
  -i reuters-seqfiles \
  -o reuters-normalized-bigram -ow \
  -a org.apache.lucene.analysis.WhitespaceAnalyzer \ # analyzer
  -chunk 200 \  # chunk size (MB)
  -wt tfidf \   # weighting scheme
  -s 5 \        # minimum support
  -md 3 \       # minimum document frequency
  -x 90 \       # maximum document frequency percentage
  -ng 2 \       # ngram size
  -ml 50 \      # minimum log likelihood ratio
  -seq \        # create sequential access sparse vectors
  -n 2          # normalization - use 2-norm (aka Euclidean norm)
                # should be paired with similar distance measure
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # two additional directories created,
hduser@cyclone:mahout$ # tokenized-documents and wordcount.
hduser@cyclone:mahout$ #######################################################

To run the K-Means cluster, here is the sequence of mahout subcommands.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # invoke kmeans
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout kmeans \
  -i reuters-vectors/tfidf-vectors \ # input
  -c reuters-initial-clusters \      # generate initial clusters
  -o reuters-kmeans-clusters         # output
  -dm org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure \
  -cd 1.0 \ # convergence threshold
  -k 20 \   # number of clusters
  -x 20 \   # max iterations
  -cl       # run clustering after iterations
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # in the output directory, there is a clusteredPoints
hduser@cyclone:mahout$ # directory, and directories corresponding to each
hduser@cyclone:mahout$ # clustering iteration named clusters-*. The last
hduser@cyclone:mahout$ # cluster is named cluster-*-final.
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # dump clusters to local filesystem
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout clusterdump \
  -dt sequencefile \                            # format: {Integer => String}
  -d reuters-vectors/dictionary.file-* \        # dictionary: {id => word}
  -i reuters-kmeans-clusters/clusters-3-final \ # input
  -o clusters.txt \                             # output (local filesystem)
  -b 10 \                                       # format length
  -n 10                                         # number of top terms to print

The dm flag specifies the Distance measure. In general, it is better to use either Cosine or Tanimoto distance measures for clustering text rather than the SquaredEuclidean measure.

The value of k (number of clusters) is provided by the caller based on his knowledge of the data. You can eliminate the guesswork by using the Canopy clusterer with appropriate distance thresholds to indicate the size of clusters as shown below.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # generate initial clusters with canopy
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout canopy \
  -i reuters-vectors/tfidf-vectors \
  -o reuters-canopy-centroids \
  -dm org.apache.mahout.common.distance.EuclideanDistanceMeasure \
  -t1 1500 \ # points between 0 and t1 from centroid are included
  -t2 2000   # points between t1 and t2 from centroid are removed
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # run kmeans with canopy cluster centroids.
hduser@cyclone:mahout$ # in this case, since -c points to a populated path,
hduser@cyclone:mahout$ # the algo doesn't have to guess initial centroids.
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout kmeans \
  -i reuters-vectors/tfidf-vectors \
  -o reuters-kmeans-clusters \
  -c reuters-canopy-centroids/clusters-0-final \
  -dm org.apache.mahout.common.distance.TanimotoDistanceMeasure \
  -cd 0.1 \     # convergence threshold
  -ow \         # overwrite
  -x 20 \       # max iterations
  -cl           # run clustering after iterations
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # view the clusters generated
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout clusterdump \
  -dt sequencefile \                     # format {Integer => String}
  -d reuters-vectors/dictionary.file-* \ # dictionary (id => word)
  -i reuters-kmeans-clusters/clusters-1-final \ # input directory
  -o clusters.txt \                      # output (local filesystem)
  -b 10 \                                # max size of word to print
  -n 10                                  # number of top terms

You can also write your own custom clusterer. I suppose this would be useful when you want to deploy an end-to-end solution but as you can see, there is not much difference from using a pipeline of commands as shown above. Here is an example of a custom clusterer that works against the Reuters dataset, using Canopy to compute the initial centroids, then using KMeans (or optionally Fuzzy KMeans) for clustering.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
// Source: :r src/main/scala/com/mycompany/mia/cluster/ReutersClusterer.scala
package com.mycompany.mia.cluster

import java.io.{StringReader, Reader}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{Path, FileSystem}
import org.apache.hadoop.io.{SequenceFile, IntWritable}
import org.apache.lucene.analysis.standard.{StandardTokenizer, StandardFilter, StandardAnalyzer}
import org.apache.lucene.analysis.tokenattributes.TermAttribute
import org.apache.lucene.analysis.{WhitespaceTokenizer, TokenStream, StopFilter, LowerCaseFilter, Analyzer}
import org.apache.lucene.util.Version
import org.apache.mahout.clustering.canopy.CanopyDriver
import org.apache.mahout.clustering.classify.WeightedVectorWritable
import org.apache.mahout.clustering.fuzzykmeans.FuzzyKMeansDriver
import org.apache.mahout.clustering.kmeans.KMeansDriver
import org.apache.mahout.clustering.Cluster
import org.apache.mahout.common.distance.{TanimotoDistanceMeasure, EuclideanDistanceMeasure}
import org.apache.mahout.common.HadoopUtil
import org.apache.mahout.vectorizer.tfidf.TFIDFConverter
import org.apache.mahout.vectorizer.{DocumentProcessor, DictionaryVectorizer}

object ReutersClusterer extends App {
  
  // parameters
  val minSupport = 2
  val minDf = 5
  val maxDfPercent = 95
  val maxNGramSize = 2
  val minLLRValue = 50
  val reduceTasks = 1
  val chunkSize = 200
  val norm = 2
  val sequentialAccessOutput = true
  
  val inputDir = args(0) // directory of doc sequence file(s)
  val outputDir = args(1) // directory where clusters will be written
  val algo = args(2) // "kmeans" or "fkmeans"
  
  val conf = new Configuration()
  val fs = FileSystem.get(conf)
  HadoopUtil.delete(conf, new Path(outputDir))
  
  // converts input docs in sequence file format in input_dir 
  // into token array in output_dir/tokenized-documents
  val inputPath = new Path(inputDir)
  val tokenizedDocPath = new Path(outputDir, 
    DocumentProcessor.TOKENIZED_DOCUMENT_OUTPUT_FOLDER) 
  DocumentProcessor.tokenizeDocuments(inputPath, 
    classOf[ReutersAnalyzer], tokenizedDocPath, conf)
    
  // reads token array in output_dir/tokenized-documents and
  // writes term frequency vectors in output_dir (under tf-vectors)
  DictionaryVectorizer.createTermFrequencyVectors(
    tokenizedDocPath,
    new Path(outputDir),
    DictionaryVectorizer.DOCUMENT_VECTOR_OUTPUT_FOLDER,
    conf, minSupport, maxNGramSize, minLLRValue, 2, true, reduceTasks,
    chunkSize, sequentialAccessOutput, false)

  // converts term frequency vectors in output_dir/tf-vectors
  // to TF-IDF vectors in output_dir (under tfidf-vectors)
  val tfVectorPath = new Path(outputDir, 
    DictionaryVectorizer.DOCUMENT_VECTOR_OUTPUT_FOLDER)
  val outputPath = new Path(outputDir)
  val docFreqs = TFIDFConverter.calculateDF(
    tfVectorPath, outputPath, conf, chunkSize)
  TFIDFConverter.processTfIdf(tfVectorPath, outputPath,
    conf, docFreqs, minDf, maxDfPercent, norm, true, 
    sequentialAccessOutput, false, reduceTasks)
    
  // reads tfidf-vectors from output_dir/tfidf-vectors
  // and writes out Canopy centroids at output_dir/canopy-centroids
  val tfidfVectorPath = new Path(outputDir, "tfidf-vectors")
  val canopyCentroidPath = new Path(outputDir, "canopy-centroids")
  CanopyDriver.run(conf, tfidfVectorPath, canopyCentroidPath,
    new EuclideanDistanceMeasure(), 250, 120, false, 0.01, false)
    
  // reads tfidf-vectors from output_dir/tfidf-vectors and 
  // refers to directory path for initial clusters, and 
  // writes out clusters to output_dir/clusters
  val clusterPath = new Path(outputDir, "clusters")
  algo match {
    case "kmeans" => KMeansDriver.run(conf, tfidfVectorPath, 
      new Path(canopyCentroidPath, "clusters-0-final"),
      clusterPath, new TanimotoDistanceMeasure(), 
      0.01, 20, true, 0.01, false)
    case "fkmeans" => FuzzyKMeansDriver.run(conf, tfidfVectorPath,
      new Path(canopyCentroidPath, "clusters-0-final"),
      clusterPath, new TanimotoDistanceMeasure(), 
      0.01, 20, 2.0f, true, true, 0.0, false)
    case _ => throw new IllegalArgumentException(
      "algo can be either kmeans or fkmeans")
  }
  
  // read clusters and output
  val reader = new SequenceFile.Reader(fs, 
    new Path(clusterPath, Cluster.CLUSTERED_POINTS_DIR + "/part-m-00000"), 
    conf)
  val key = new IntWritable()
  val value = new WeightedVectorWritable()
  while (reader.next(key, value)) {
    println(key.toString + " belongs to " + value.toString)
  }
  reader.close()
}

class ReutersAnalyzer extends Analyzer {
  
  val ALPHA_PATTERN = """[a-z]+""".r
  
  override def tokenStream(fieldName : String, reader : Reader) : 
   TokenStream = {
    // tokenize input string by standard tokenizer
    var result : TokenStream = 
      new StandardTokenizer(Version.LUCENE_CURRENT, reader)
    result = new StandardFilter(result)
    // lowercase all words
    result = new LowerCaseFilter(result)
    // remove stop words
    result = new StopFilter(true, result, 
      StandardAnalyzer.STOP_WORDS_SET)
    val termAttr = result.addAttribute(classOf[TermAttribute]).
      asInstanceOf[TermAttribute]
    val buf = new StringBuilder()
    while (result.incrementToken()) {
      // remove words < 3 chars long
      if (termAttr.termLength() >= 3) {
        val word = new String(
          termAttr.termBuffer(), 0, termAttr.termLength())
        // remove words with non-alpha chars in them
        if (ALPHA_PATTERN.pattern.matcher(word).matches) {
          buf.append(word).append(" ")
        }
      }
    }
    // return the remaining tokens
    new WhitespaceTokenizer(new StringReader(buf.toString))
  }
}

Just like K-Means, the other clustering algorithms can also be run either form a script or via code as shown above. Here are the commands for the other major clustering algorithms.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # run fuzzy kmeans with random initial centroids
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout fkmeans \
  -i reuters-vectors/tfidf-vectors \
  -c reuters-fkmeans-centroids \
  -o reuters-fkmeans-clusters \
  -cd 1.0 \ # convergence threshold
  -k 21 \
  -ow \
  -x 10 \ # maxiterations
  -m 2.0 \ # fuzzification factor
  -dm org.apache.mahout.common.distance.SquaredEuclideanDistanceMeasure
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # view generated clusters.
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout clusterdump \
  -dt sequencefile \
  -d reuters-vectors/dictionary.file-* \
  -i reuters-fkmeans-clusters/clusters-2-final \
  -o clusters.txt \
  -b 10 \
  -n 10

Dirichlet clustering is based on a probabilistic model, that attempts to predict the distribution of points around a cluster. Model Distributions provided by Mahout are the DistanceMeasureClusterDistribution and GaussianClusterDistribution, and of course users can implement ModelDistribution to create their own custom one as well. The command line invocation of Dirichlet clusterer is shown below:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # invoke dirichlet clustering
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout dirichlet \
  -i reuters-vectors/tfidf-vectors \
  -o reuters-dirichlet-clusters \
  -k 60 \                          # number of clusters
  -x 10 \                          # number of iterations
  -a0 1.0 \                        # alpha0 value
  -md org.apache.mahout.clustering.dirichlet.models.GaussianClusterDistribution \
  -mp org.apache.mahout.math.SequentialAccessSparseVector #default vector type
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ # view generated clusters.
hduser@cyclone:mahout$ #######################################################
hduser@cyclone:mahout$ bin/mahout clusterdump \
  -dt sequencefile \
  -d reuters-vectors/dictionary.file-* \
  -i reuters-dirichlet-clusters/clusters-0 \
  -o clusters.txt \
  -b 10 \
  -n 10

The book covers another algorithm called LDA (Latent Dirichlet Allocation) which is useful for Topic Modeling, but the subcommand has been deprecated in version 0.7 and replaced with two other subcommands which I could not make working. I guess I will come back to this one later when I actually need it, and when I am a bit more familiar with Mahout.

Case Study - LastFM

The Last.fm dataset contains (music) group names tagged with words or phrases and the respective tag counts. The objective is to suggest tags for artists given their existing tags, ie to cluster tags using the artists as feature vector and tag counts as feature weights. Unlike the Reuters case, the approach here is to write just enough code to generate the dictionary (for clusterdump) and feature vectors (for the clusterer). For the actual clustering and viewing the clusters, we depend on the appropriate Mahout subcommands. Here is the code for the vectorizer.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
// Source: src/main/scala/com/mycompany/mia/cluster/LastFMVectorizer.scala
package com.mycompany.mia.cluster

import scala.collection.JavaConversions.iterableAsScalaIterable

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{Path, FileSystem}
import org.apache.hadoop.io.{IntWritable, DefaultStringifier, Text, SequenceFile, LongWritable}
import org.apache.hadoop.mapreduce.lib.input.{FileInputFormat, TextInputFormat}
import org.apache.hadoop.mapreduce.lib.output.{FileOutputFormat, SequenceFileOutputFormat}
import org.apache.hadoop.mapreduce.{Mapper, Reducer, Job}
import org.apache.hadoop.util.{GenericsUtil, GenericOptionsParser}
import org.apache.mahout.math.{VectorWritable, Vector, SequentialAccessSparseVector, NamedVector}

/**
 * We need to read a file of the following format:
 * UUID<sep>Artist Name<sep>Tag<sep>Count
 * into a sequence file of [Text,VectorWritable] as follows:
 * Tag => VectorWritable(Artist:Count)
 */
object LastFMVectorizer {

  def main(args : Array[String]) : Int = {
    val conf = new Configuration()

    val otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs
    if (otherArgs.length != 2) {
      println("Usage: LastFMVectorizer input_file output_dir")
      -1
    }

    // Dictionary Mapper/Reducer. Extract unique artists
    val job1 = new Job(conf, "Dictionary Mapper")
    job1.setJarByClass(classOf[DictionaryMapper])
    job1.setMapperClass(classOf[DictionaryMapper])
    job1.setReducerClass(classOf[DictionaryReducer])
    job1.setOutputKeyClass(classOf[Text])
    job1.setOutputValueClass(classOf[IntWritable])
    job1.setInputFormatClass(classOf[TextInputFormat])
    job1.setOutputFormatClass(classOf[SequenceFileOutputFormat[Text,IntWritable]])
    FileInputFormat.addInputPath(job1, new Path(args(0)))
    val dictOutput = new Path(args(1), "dictionary")
    FileOutputFormat.setOutputPath(job1, dictOutput)
    var succ = (job1.waitForCompletion(true))
    
    if (succ) {
      // get a mapping of unique ids to artist and tag for converting
      // to Mahout vectors
      val dictOutput = new Path(args(1), "dictionary")
      val fs = FileSystem.get(dictOutput.toUri(), conf)
      val dictfiles = fs.globStatus(new Path(dictOutput, "part-*"))
      var i = 0
      val dictGlob = new Path(args(1), "dict-glob")
      val writer = new SequenceFile.Writer(fs, conf, dictGlob, 
        classOf[Text], classOf[IntWritable])
      for (dictfile <- dictfiles) {
        val path = dictfile.getPath()
        val reader = new SequenceFile.Reader(fs, path, conf)
        val key = new Text()
        val value = new IntWritable()
        while (reader.next(key, value)) {
          writer.append(key, new IntWritable(i))
          i += 1
        }
        reader.close()
      }
      writer.close()
      conf.set("dictpath", dictGlob.toString())

      val job2 = new Job(conf, "Dictionary Vectorizer")
      job2.setJarByClass(classOf[VectorMapper])
      job2.setMapperClass(classOf[VectorMapper])
      job2.setReducerClass(classOf[VectorReducer])
      job2.setOutputKeyClass(classOf[Text])
      job2.setOutputValueClass(classOf[VectorWritable])
      job2.setInputFormatClass(classOf[TextInputFormat])
      job2.setOutputFormatClass(classOf[SequenceFileOutputFormat[Text,VectorWritable]])
      FileInputFormat.addInputPath(job2, new Path(args(0)))
      FileOutputFormat.setOutputPath(job2, new Path(args(1), "vectors"))
      succ = (job2.waitForCompletion(true))
    }
    if (succ) 0 else 1
  }
}

/////////////////////////////////////////////////////////////////
// Assigns a unique ID to each artist. Needed by clusterdump
/////////////////////////////////////////////////////////////////

class DictionaryMapper 
    extends Mapper[LongWritable,Text,Text,IntWritable] {
  
  val pattern = """<sep>""".r
  val zero = new IntWritable(0)
  
  override def map(key : LongWritable, 
      value : Text, 
      context : Mapper[LongWritable,Text,Text,IntWritable]#Context) = { 
    val fields = pattern.split(value.toString)
    if (fields.length != 4) {
      context.getCounter("Map", "LinesWithErrors").increment(1)
    } else {
      context.write(new Text(fields(1)), zero)
    }
  }
}

class DictionaryReducer
    extends Reducer[Text,IntWritable,Text,IntWritable] {

  val zero = new IntWritable(0)

  override def reduce(key : Text, 
      values : java.lang.Iterable[IntWritable],
      context : Reducer[Text,IntWritable,Text,IntWritable]#Context) = {
    context.write(key, zero)
  }
}

/////////////////////////////////////////////////////////////////
// For each tag, creates feature vectors of artists
/////////////////////////////////////////////////////////////////

class VectorMapper
    extends Mapper[LongWritable,Text,Text,VectorWritable] {
  
  val pattern = """<sep>""".r
  var dict = new java.util.HashMap[String,Integer]()
  var vecwritable = new VectorWritable()
  
  override def setup(
      context : Mapper[LongWritable,Text,Text,VectorWritable]#Context) = {
    super.setup(context)
    val conf = context.getConfiguration()
    val dictpath = new Path(conf.get("dictpath"))
    val fs = FileSystem.get(dictpath.toUri(), conf)
    val reader = new SequenceFile.Reader(fs, dictpath, conf)
    val key = new Text()
    val value = new IntWritable()
    while (reader.next(key, value)) {
      dict.put(key.toString(), value.get())
    }
  }
  
  override def map(key : LongWritable, 
      value : Text,
      context : Mapper[LongWritable,Text,Text,VectorWritable]#Context) = {
    val fields = pattern.split(value.toString)
    if (fields.length != 4) {
      context.getCounter("Map", "LinesWithErrors").increment(1)
    } else {
      val artist = fields(1)
      val tag = fields(2)
      val weight = java.lang.Double.parseDouble(fields(3))
      val vector = new NamedVector(
        new SequentialAccessSparseVector(dict.size()), tag)
      vector.set(dict.get(artist), weight)
      vecwritable.set(vector)
      context.write(new Text(tag), vecwritable)
    }
  }
}

class VectorReducer
    extends Reducer[Text,VectorWritable,Text,VectorWritable] {
  
  var vecwritable = new VectorWritable()
  
  override def reduce(key : Text,
      values : java.lang.Iterable[VectorWritable],
      context : Reducer[Text,VectorWritable,Text,VectorWritable]#Context) = {
    var vector : Vector = null
    for (partialVector <- values) {
      if (vector == null) {
        vector = partialVector.get().like()
      } else {
        vector.plus(partialVector.get())
      }
    }
    val artistVector = new NamedVector(vector, key.toString())
    vecwritable.set(artistVector)
    context.write(key, vecwritable)
    for (value <- values) {
      context.write(key, value)
    }
  }
}

Once the vectors are generated, we just use Mahout subcommands to complete the clustering work.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
hduser@cyclone:mahout$ hadoop jar /tmp/my-mahout-fatjar.jar \
  com.mycompany.mia.cluster.LastFMVectorizer \
  input/ArtistTags.dat \
  output
hduser@cyclone:mahout$ bin/mahout kmeans \
  -i output/vectors \
  -o output/topics \
  -c output/centroids \
  -k 200 \
  -ow \
  -dm org.apache.mahout.common.distance.CosineDistanceMeasure \
  -cd 0.01 \
  -x 20 \
  -cl
hduser@cyclone:mahout$ bin/mahout clusterdump \
  -i output/topics/clusters-20-final \
  -d output/dict-glob \
  -dt sequencefile \
  -n 10 \
  -o clusters.txt

There are two other case studies in the MIA book which I didn't do, maybe I will come back and do them once I finish the other stuff in the book. I also skipped over LDA (because its now deprecated in favor of two other alogorithms and I couldn't make either of them work). However, I now have a decent understanding of how to do clustering with Mahout.

If you've been reading this post with the intent of learning about Mahout clustering, you may have found the pace a little too fast. Thats because there is a lot of material to cover in a relatively short post. For a gentler introduction please refer to the MIA book and the companion code repository on GitHub..

Posted by Sujit Pal at 12:55 PM 58 comments

Labels: , , ,

Share/Save: Digg Technorati Delicious StumbleUpon Reddit BlinkList Furl Mixx Facebook Google Bookmark Yahoo

Subscribe to: Posts (Atom)