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