KmeansAdapter: Hadoop K-Means 聚类算法适配器

package com.huiluczP.cluster;

import com.huiluczP.KmeansRun; import com.huiluczP.util.CalUtil; import com.huiluczP.util.DataUtil; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import java.io.IOException; import java.util.ArrayList;

public class KmeansAdapter {

public static boolean checkStop(String centerPath, String newCenterPath){
    try {
        ArrayList<ArrayList<Double>> newCenters = DataUtil.readCenter(newCenterPath);
        ArrayList<ArrayList<Double>> centers = DataUtil.readCenter(centerPath);
        double distanceSum = CalUtil.calDistanceBetweenCenters(centers, newCenters);
        if(distanceSum == 0){
            return true;
        }else{
            System.out.println('distanceSum=' + distanceSum);
            DataUtil.changeCenters(centerPath, newCenterPath, new Configuration());
            return false;
        }
    } catch (IOException e) {
        System.out.println(centerPath + ' ! file wrong');
        e.printStackTrace();
    }
    return true;
}

public static void start(String dataPath, String centerPath, String newCenterPath){
    Configuration hadoopConfig = new Configuration();
    hadoopConfig.set('cluster.center_path', centerPath);

    try {
        Job job = Job.getInstance(hadoopConfig, 'one round cluster task');

        job.setJarByClass(KmeansRun.class);
        job.setMapperClass(KmeansMapper.class);
        job.setReducerClass(KmeansReducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        Path outPath = new Path(newCenterPath);
        outPath.getFileSystem(hadoopConfig).delete(outPath, true);

        FileInputFormat.addInputPath(job, new Path(dataPath));
        FileOutputFormat.setOutputPath(job, new Path(newCenterPath));

        job.waitForCompletion(true);
        System.out.println('finish one round cluster task');
    } catch (IOException | InterruptedException | ClassNotFoundException e) {
        e.printStackTrace();
    }
}

public static void createClusterResult(String dataPath, String centerPath, String clusterResultPath){
    Configuration hadoopConfig = new Configuration();
    hadoopConfig.set('cluster.center_path', centerPath);

    try {
        Job job = Job.getInstance(hadoopConfig, 'cluster result task');

        job.setJarByClass(KmeansRun.class);
        job.setMapperClass(KmeansMapper.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        Path outPath = new Path(clusterResultPath);
        outPath.getFileSystem(hadoopConfig).delete(outPath, true);

        FileInputFormat.addInputPath(job, new Path(dataPath));
        FileOutputFormat.setOutputPath(job, new Path(clusterResultPath));

        job.waitForCompletion(true);
        System.out.println('cluster result task finished');

    } catch (IOException | InterruptedException | ClassNotFoundException e) {
        e.printStackTrace();
    }
}

}