zoukankan      html  css  js  c++  java

  • 【转】mapreduce中的全局文件使用方法：以k-means为例

    【转自：http://www.linuxidc.com/Linux/2012-10/71540p3.htm】　　

    以前有做过在Hadoop编 写程序时使用全局变量的想法，但是最后却没有实现，上网查才看到说Hadoop不支持全局变量。但是有时候编程的时候又会用到，比如编写k-means算 法的时候，如果可以有个全局变量存储中心点该多好呀。其实在hadoop中确实是有相关的实现的，比如可以在mapper中的setup函数中读取一个小 文件，然后从这个文件中取出全局变量的值。

    那具体如何实现呢？首先提出一个问题，然后利用这种思想去解决会比较好。首先说下我要实现的问题：我现在有输入数据如下：

    1. 0.0 0.2 0.4 
    2. 0.3 0.2 0.4 
    3. 0.4 0.2 0.4 
    4. 0.5 0.2 0.4 
    5. 5.0 5.2 5.4 
    6. 6.0 5.2 6.4 
    7. 4.0 5.2 4.4 
    8. 10.3    10.4    10.5 
    9. 10.3    10.4    10.5 
    10. 10.3    10.4    10.5 

    而且还有一个小数据文件（中心点）如下：

    1. 0   0   0 
    2. 5   5   5 
    3. 10  10  10 

    我想做的事情就是把输入数据按照中心点求平均值，即首先我把输入数据分类，比如倒数三行应该都是属于（10，10，10）这个中心点的，那么我的 map就把倒数三行的key都赋值为2，然后value值还是保持这三行不变。在reduce阶段，我求出相同key的sum值，同时求出一共的行数 count，最后我用sum/count得到我想要的按照中心点求出的平均值了。

    下面贴代码：

    KmeansDriver:

    1. package org.fansy.date927; 
    2.  
    3. import java.io.IOException; 
    4.  
    5. //import org.apache.commons.logging.Log;  
    6. //import org.apache.commons.logging.LogFactory;  
    7. import org.apache.hadoop.conf.Configuration; 
    8. import org.apache.hadoop.filecache.DistributedCache; 
    9. //import org.apache.hadoop.fs.FileSystem;  
    10. import org.apache.hadoop.fs.Path; 
    11. import org.apache.hadoop.io.IntWritable; 
    12. import org.apache.hadoop.io.NullWritable; 
    13. import org.apache.hadoop.io.Text; 
    14. import org.apache.hadoop.mapreduce.Job; 
    15. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
    16. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
    17. import org.apache.hadoop.util.GenericOptionsParser; 
    18.  
    19. public class KmeansDriver { 
    20.     /** 
    21.      *   k-means algorithm program   
    22.      */ 
    23.     public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException { 
    24.         // TODO Auto-generated method stub  
    25.         Configuration conf=new Configuration(); 
    26.         // set the centers data file  
    27.         <SPAN style="COLOR: #ff0000">Path centersFile=new Path("hdfs://fansyPC:9000/user/fansy/input/centers"); 
    28.         DistributedCache.addCacheFile(centersFile.toUri(), conf);</SPAN> 
    29.         String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs(); 
    30.         if (otherArgs.length != 2) { 
    31.           System.err.println("Usage: KmeansDriver <in> <out>"); 
    32.           System.exit(2); 
    33.         } 
    34.         Job job = new Job(conf, "kmeans job"); 
    35.         job.setJarByClass(KmeansDriver.class); 
    36.         job.setMapperClass(KmeansM.class); 
    37.         job.setMapOutputKeyClass(IntWritable.class); 
    38.         job.setMapOutputValueClass(DataPro.class); 
    39.         job.setNumReduceTasks(2); 
    40.                 //    job.setCombinerClass(KmeansC.class);  
    41.         job.setReducerClass(KmeansR.class); 
    42.         job.setOutputKeyClass(NullWritable.class); 
    43.         job.setOutputValueClass(Text.class); 
    44.          
    45.         FileInputFormat.addInputPath(job, new Path(otherArgs[0])); 
    46.         FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); 
    47.          
    48.         if(!job.waitForCompletion(true)){ 
    49.             System.exit(1); // run error then exit  
    50.         }     
    51.     } 
    52.  
    53. }
    55. Mapper:
      1. package org.fansy.date927; 
      2.  
      3. import java.io.BufferedReader; 
      4. import java.io.FileReader; 
      5. import java.io.IOException; 
      6. import java.util.ArrayList; 
      7. import java.util.List; 
      8.  
      9. import org.apache.commons.logging.Log; 
      10. import org.apache.commons.logging.LogFactory; 
      11. import org.apache.Hadoop.filecache.DistributedCache; 
      12. import org.apache.hadoop.fs.Path; 
      13. import org.apache.hadoop.io.IntWritable; 
      14. import org.apache.hadoop.io.LongWritable; 
      15. import org.apache.hadoop.io.Text; 
      16. import org.apache.hadoop.mapreduce.Mapper; 
      17.  
      18. public class KmeansM extends Mapper<LongWritable,Text,IntWritable,DataPro>{ 
      19.     private static Log log=LogFactory.getLog(KmeansM.class); 
      20.      
      21.     private double[][] centers; 
      22.     private int dimention_m;  //  this is the k   
      23.     private int dimention_n;   //  this is the features   
      24.  
      25.      
      26.     static enum Counter{Fansy_Miss_Records}; 
      27.     @Override 
      28.     public void setup(Context context) throws IOException,InterruptedException{ 
      29.         <SPAN style="COLOR: #ff0000">Path[] caches=DistributedCache.getLocalCacheFiles(context.getConfiguration());</SPAN> 
      30.         if(caches==null||caches.length<=0){ 
      31.             log.error("center file does not exist"); 
      32.             System.exit(1); 
      33.         } 
      34.         @SuppressWarnings("resource") 
      35.         BufferedReader br=new BufferedReader(new FileReader(caches[0].toString())); 
      36.         String line; 
      37.         List<ArrayList<Double>> temp_centers=new ArrayList<ArrayList<Double>>(); 
      38.         ArrayList<Double> center=null; 
      39.         //  get the file data  
      40.         <SPAN style="COLOR: #ff0000">while((line=br.readLine())!=null){ 
      41.             center=new ArrayList<Double>(); 
      42.             String[] str=line.split(" "); 
      43.             for(int i=0;i<str.length;i++){ 
      44.                 center.add(Double.parseDouble(str[i])); 
      45.             } 
      46.             temp_centers.add(center); 
      47.         }</SPAN> 
      48.         //  fill the centers   
      49.         @SuppressWarnings("unchecked") 
      50.         ArrayList<Double>[] newcenters=temp_centers.toArray(new ArrayList[]{}); 
      51.          dimention_m=temp_centers.size(); 
      52.          dimention_n=newcenters[0].size(); 
      53.         centers=new double[dimention_m][dimention_n]; 
      54.         for(int i=0;i<dimention_m;i++){ 
      55.             Double[] temp_double=newcenters[i].toArray(new Double[]{}); 
      56.             for(int j=0;j<dimention_n;j++){ 
      57.                 centers[i][j]=temp_double[j]; 
      58.         //      System.out.print(temp_double[j]+",");  
      59.             } 
      60.     //      System.out.println();  
      61.         } 
      62.     } 
      63.              
      64.     public void map(LongWritable key,Text value,Context context)throws IOException,InterruptedException{ 
      65.         String[] values=value.toString().split(" "); 
      66.          
      67.         if(values.length!=dimention_n){ 
      68.             context.getCounter(Counter.Fansy_Miss_Records).increment(1); 
      69.             return; 
      70.         } 
      71.         double[] temp_double=new double[values.length]; 
      72.         for(int i=0;i<values.length;i++){ 
      73.             temp_double[i]=Double.parseDouble(values[i]); 
      74.         } 
      75.         //  set the index  
      76.         double distance=Double.MAX_VALUE; 
      77.         double temp_distance=0.0; 
      78.         int index=0; 
      79.         for(int i=0;i<dimention_m;i++){ 
      80.             double[] temp_center=centers[i]; 
      81.             temp_distance=getEnumDistance(temp_double,temp_center); 
      82.             if(temp_distance<distance){ 
      83.                  index=i; 
      84.                 distance=temp_distance; 
      85.             } 
      86.         } 
      87.         DataPro newvalue=new DataPro(); 
      88.         newvalue.set(value, new IntWritable(1)); 
      89.         context.write(new IntWritable(index), newvalue); 
      90.          
      91.     } 
      92.     public static double getEnumDistance(double[] source,double[] other){  //  get the distance  
      93.         double distance=0.0; 
      94.         if(source.length!=other.length){ 
      95.             return Double.MAX_VALUE; 
      96.         } 
      97.         for(int i=0;i<source.length;i++){ 
      98.             distance+=(source[i]-other[i])*(source[i]-other[i]); 
      99.         } 
      100.         distance=Math.sqrt(distance); 
      101.         return distance; 
      102.     } 
      103. } 

      Reducer:

      1. package org.fansy.date927; 
      2.  
      3. import java.io.IOException; 
      4. import org.apache.Hadoop.io.IntWritable; 
      5. import org.apache.hadoop.io.NullWritable; 
      6. import org.apache.hadoop.io.Text; 
      7. import org.apache.hadoop.mapreduce.Reducer; 
      8.  
      9. public class KmeansR extends Reducer<IntWritable,DataPro,NullWritable,Text> { 
      10.      
      11.     public void reduce(IntWritable key,Iterable<DataPro> values,Context context)throws InterruptedException, IOException{ 
      12.         // get dimension first  
      13.         int dimension=0; 
      14.         for(DataPro val:values){ 
      15.             String[] datastr=val.getCenter().toString().split(" "); 
      16.             dimension=datastr.length; 
      17.             break; 
      18.         } 
      19.         double[] sum=new double[dimension]; 
      20.         int sumCount=0; 
      21.         for(DataPro val:values){ 
      22.             String[] datastr=val.getCenter().toString().split(" "); 
      23.             sumCount+=val.getCount().get(); 
      24.             for(int i=0;i<dimension;i++){ 
      25.                 sum[i]+=Double.parseDouble(datastr[i]); 
      26.             } 
      27.         } 
      28.         //  calculate the new centers  
      29. //      double[] newcenter=new double[dimension];  
      30.         StringBuffer sb=new StringBuffer(); 
      31.         for(int i=0;i<dimension;i++){ 
      32.             sb.append(sum[i]/sumCount+" "); 
      33.         } 
      34.         context.write(null, new Text(sb.toString())); 
      35.     } 
      36. } 

      DataPro:

      1. package org.fansy.date927; 
      2.  
      3. import java.io.DataInput; 
      4. import java.io.DataOutput; 
      5. import java.io.IOException; 
      6.  
      7. import org.apache.hadoop.io.IntWritable; 
      8. import org.apache.hadoop.io.Text; 
      9. import org.apache.hadoop.io.WritableComparable; 
      10.  
      11. public class DataPro implements WritableComparable<DataPro>{ 
      12.  
      13.     private Text center; 
      14.     private IntWritable count; 
      15.      
      16.     public DataPro(){ 
      17.         set(new Text(),new IntWritable()); 
      18.     } 
      19.     public void set(Text text, IntWritable intWritable) { 
      20.         // TODO Auto-generated method stub  
      21.         this.center=text; 
      22.         this.count=intWritable; 
      23.     } 
      24.      
      25.     public Text getCenter(){ 
      26.         return center; 
      27.     } 
      28.     public IntWritable getCount(){ 
      29.         return count; 
      30.     } 
      31.      
      32.      
      33.     @Override 
      34.     public void readFields(DataInput arg0) throws IOException { 
      35.         // TODO Auto-generated method stub  
      36.         center.readFields(arg0); 
      37.         count.readFields(arg0); 
      38.     } 
      39.  
      40.     @Override 
      41.     public void write(DataOutput arg0) throws IOException { 
      42.         // TODO Auto-generated method stub  
      43.         center.write(arg0); 
      44.         count.write(arg0); 
      45.     } 
      46.  
      47.     @Override 
      48.     public int compareTo(DataPro o) { 
      49.         // TODO Auto-generated method stub  
      50.         int cmp=count.compareTo(o.count); 
      51.         if(cmp!=0){ 
      52.             return cmp; 
      53.         } 
      54.         return center.compareTo(o.center); 
      55.     } 
      56.  
      57. } 

      这里自定义了一个DataPro数据类型，主要是为了为以后编写真正的k-means算法时使用combiner做准备，具体思想可以参考上篇combine操作。

      输出文件如下：

      1. 0.39999999999999997 0.20000000000000004 0.4000000000000001   
      2. 5.0 5.2 5.4   
      3. 10.3    10.4    10.5   

      这篇文章参考了 http://www.linuxidc.com/Linux/2012-10/71538.htm 部分实现，在那篇文章中的k-means思想的主要思想是：使用map读入centers文件值，然后把数据文件data作为一个全局量，然后reduce在进行求中心点的操作。（或许我理解错了也说不定）

      做完这一步后，如果要编写K-means算法就可以说是已经实现了大半了，剩下的就是设置下输入和输出路径，然后进行迭代了。

      【转自：http://www.linuxidc.com/Linux/2012-10/71540p3.htm】
  • 相关阅读:
    Android View体系（七）从源码解析View的measure流程
    Android View体系（六）从源码解析Activity的构成
    Android View体系（五）从源码解析View的事件分发机制
    Android View体系（四）从源码解析Scroller
    Android常用学习网站
    Android View体系（三）属性动画
    购物车
    模块与包
    s5_day9作业
    s5_day11作业
  • 原文地址：https://www.cnblogs.com/aitixiaocai/p/3561935.html
Copyright © 2011-2022 走看看