mahout-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Angelo Immediata <angelo...@gmail.com>
Subject Mahout KMeans: is this the correct way?
Date Thu, 12 Dec 2013 10:42:57 GMT
Hi
I'm using this environment:

   - Apache Hadoop 1.2.1
   - Apache HBase  0.96 hadoop1
   - Apache Mahout 0.8
   - Spring 3.2

I have to do some clustering on records stored in my HBase table. Records
have these properties:

   - idArco: the arc ID of the route I'm considering
   - velocitaMedia: medium velocity on that route
   - matchingQuality: quality of the misuration
   - startDate: misuration start date
   - endDate: misuration start date
   - vehiclesNumber: number of vehicles involved in the misuration
   - meseAnno: misuration month of the year
   - giornoSettimana: week day of the misuration
   - oraGiorno: day hour of the misuration
   - calendarioFestivo: numeric representing if there has been some kind of
   festivity during the misuration
   - calendarioEventi: numeric representing if there has been some kind of
   event during the misuration
   - eventoMeteo: numeric representing the weather situation during the
   misuration
   - manifestazione: numeric representing if there has been some kind of
   manifestation during the misuration
   - annoMisurazione: year of the misuration
   - tipoStrada: route tipology (if it's highway, normal way and so on...)
   - minutoOra: minutes of the misuration hour
   - startDateLong: misuration start date in millis
   - endDateLong: misuration final date in millis
   - idCluster: record membership cluster id

I need to realize the cluster according to the following properties of my
records: calendarioFestivo, calendarioEventi, eventoMeteo, manifestazione
(and maybe some othe properties like oraGiorno, minutoOra and so on...but
for now it's good to stop here; if I did my work correctly I guess it will
be simple to add some other parameters); the final result should be
something like this:

   - cluster 0 is the cluster for a well known weather condition, with a
   well known kind of festivity and event and so on...
   - cluster 1 is the cluster for another well known weather condition,
   with a well known kind of festivity and event and so on...

In order to realize the clustering I'm using Mahout KMeans Implementation
and in order to find the right K value I used Canopy.
The first thing I did is to take all data from my HBase by using the HBase
MapReduce funcionality; by all these data I wrote the SequenceFile
representing the input data for the clustering algorithm, so I wrote this
code:
//*Map side*:
protected void map(ImmutableBytesWritable key, Result result,
org.apache.hadoop.mapreduce.Mapper.Context context) throws IOException,
InterruptedException {
try{
Double calFest = //Taken from DB
Double calEven = //Taken from DB
Double meteo = //Taken from DB
Double manifestazione = //Taken from DB
Double tipologiaStrada = //Taken from DB
String chiave = Bytes.toString(result.getRow());
Text text = new Text();
text.set(chiave);
DenseVector dv = new DenseVector(new double[]{calFest, calEven, meteo,
manifestazione, tipologiaStrada});
NamedVector nv = new NamedVector(dv, chiave);
context.write(text, new VectorWritable(nv));
}catch(Exception e){
//Handled e
}
}
//*Reduce side*
    public static class HistoricalDataReducer extends Reducer<Text,
VectorWritable, Text, VectorWritable> {
        private SequenceFile.Writer sfWriter;
        protected void cleanup(Context context) throws IOException,
InterruptedException {
            try {
                sfWriter.close();
            } catch (Exception e) {
              //Handled e
            }
        }
        protected void setup(Context context) throws IOException,
InterruptedException {
            try {
                Path inputDataSeq = new Path("somePath");
                Configuration conf = new Configuration();
                HadoopUtil.delete(conf, inputDataSeq);
                FileSystem fs = FileSystem.get(conf);
                this.sfWriter = new SequenceFile.Writer(fs, conf,
inputDataSeq, Text.class, VectorWritable.class);
            } catch (Exception e) {
                //Handled e
            }
        }
        protected void reduce(Text key, Iterable<VectorWritable> values,
Context context) throws IOException, InterruptedException {
            try{
                Iterator<VectorWritable> iterator = values.iterator();
                while (iterator.hasNext()) {
                    VectorWritable vectorWritable = iterator.next();
                    NamedVector nv = (NamedVector)vectorWritable.get();
                    this.sfWriter.append(new Text( nv.getName()),
vectorWritable);
                    context.write(key, vectorWritable);
                }
            }catch(Exception e){
                //Handled e
            }
        }
        }

Then I wrote this class for the cluster analysis:
//Import stuff
@Service
public class ClusterAnalysisSvcImpl implements IClusterAnalysisSvc {
//Some configuration properties
@Override
public void executeClusterAnalysis() throws ClusterAnalysisException {

try {
DistanceMeasure measure = new EuclideanDistanceMeasure();
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(conf);
writeInputData(fs, conf);
writeClusters(fs, conf, measure);
Path inputData = new Path(inputDataLocation);
Path inputClaster = new Path(inputClusterDirectory);
Path outputResults = new Path(outputClusterDirectory);
HadoopUtil.delete(conf, outputResults);
sw.start("fase di clustering");
KMeansDriver.run(conf,
inputData,
inputClaster,
outputResults,
measure,
convergenceDelta,
iterationNumber,
true,
0,
false);
} catch (Exception e) {
//Handled e
}
}
private List<Canopy> getCanopies(FileSystem fs, Configuration conf,
DistanceMeasure measure) throws Exception{
SequenceFile.Reader reader = null;
try{
reader = new SequenceFile.Reader(fs, new Path(inputDataLocation), conf);
Text key = new Text();
VectorWritable value = new VectorWritable();
List<Vector> elementi = new ArrayList<Vector>();
while(reader.next(key, value)){
NamedVector nv = (NamedVector)value.get();
elementi.add(nv);
}
if( t1<=t2 ){
throw new IllegalArgumentException("Impossibile proseguire; le soglie di
distanza Canopy non sono corrette; t1 รจ minore o uguale a t2: t1="+t1+"
t2="+t2);
}
//Individuo i canopy
List<Canopy> canopies = CanopyClusterer.createCanopies(elementi, measure,
t1, t2);
return canopies;
}finally{
if( reader != null ){
reader.close();
}
}
}
private void writeClusters(FileSystem fs, Configuration conf,
DistanceMeasure measure) throws Exception{
SequenceFile.Writer writer = null;
try {
Path path = new Path(clusterPathFile);
HadoopUtil.delete(conf, path);
List<Canopy> canopies = getCanopies(fs, conf, measure);
writer = new SequenceFile.Writer(fs, conf, path, Text.class, Kluster.class);
for (Canopy canopy : canopies) {
Kluster clust = new Kluster(canopy.getCenter(), canopy.getId(), measure);
writer.append(new Text( (clust.getIdentifier()) ), clust);
}
}finally {
if( writer != null ){
writer.close();
}
}
}
private void writeInputData(FileSystem fs, Configuration conf) throws
Exception{
try {
//Calling the HBase MapReduce utility
}
}

In this way, by using a dataset of 28800 records, I found 360 clusters.
Since I'm newbie I'ld like to know what you, experts, think about my
approach; is this a good way to realize the cluster analysis? Or should I
do it in other way?

Thank you

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message