项目组hadoop实战之归一化：每天更新中。。。

pig2 发表于 2015-1-30 15:51
伪代码如下：
int key=x;
int b=y;

版主，怪我没说清楚哈。我更正了一下，麻烦再看一下哈

s060403072 发表于 2015-1-30 17:11
你的需求需要说的详细些，人家的方法已经说的很明白了，即使不一样，你根据这个改改就可以了。

嗯，怪我表述的不细致。已经重新再次编辑了哈。麻烦再次详阅哈。

Hedera 发表于 2015-1-30 18:22
嗯，怪我表述的不细致。已经重新再次编辑了哈。麻烦再次详阅哈。

比例还是不太明白，这个比例是否固定，你的思路不是已经被否定了吗

s060403072 发表于 2015-1-30 19:28
比例还是不太明白，这个比例是否固定，你的思路不是已经被否定了吗

比如有txt的数据如下：
11
54
23
92
104
993
21
42

则归一化的过程如下：
找出最大993，最小11

使用归一化公式（线性函数转换）：y=(x-MinValue)/(MaxValue-MinValue)
依次计算每个归一化的结果：
(11-11)/(993-11)=0/982
(54-11)/(993-11)=43/982
(23-11)/(993-11)=12/982
(92-11)/(993-11)=81/982
……
就是这个意思，可以明白不?

Hedera 发表于 2015-1-30 19:45
比如有txt的数据如下：
11
54

你的比例是固定，还是不固定，只需要说明白这点即可，否则无论是公式，还是规律，都的不出来。

s060403072 发表于 2015-1-30 20:58
你的比例是固定，还是不固定，只需要说明白这点即可，否则无论是公式，还是规律，都的不出来。

刚问了一下，不用考虑比例固定，就是直接按归一化公式：(x-min)/(max-min)就可以了。

设计思路二：

刚把这个问题解决了，拿来和大家分享一下。

分两个job来完成。

第一个job来查找最大值最小值，并且保存在hdfs上

第二个job在map阶段使用setup函数从hdfs中读取最大值最小值来初始化。然后使用map与reduce来完成归一化

即用两个线性的mapreduce流来完成。

以下是设计思路二的代码：

package normalizedVersion1;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import readDataFromHdfs.readDataFromHdfs;

public class normalizedVersion1 {
        
        static float max;
        static float min;

  public static class findMaxMinMapper
       extends Mapper<Object, Text, Text, IntWritable>{
    
    public void map(Object key, Text value, Context context
                    ) throws IOException, InterruptedException {
      StringTokenizer itr = new StringTokenizer(value.toString());
      
      String headStr=itr.nextToken().toString();
      
      float headFloat=Float.parseFloat(headStr);
      
      context.write(new Text("flag"), new IntWritable((int) headFloat));

    }

  }

  public static class findMaxMinReducer
       extends Reducer<Text,IntWritable,Text,IntWritable> {
    private IntWritable result = new IntWritable(1);

    public void reduce(Text key, Iterable<IntWritable> values,
                       Context context
                       ) throws IOException, InterruptedException {
            int minTemp=65533,maxTemp=0;
      for (IntWritable val : values) {
        if(val.get()<(int)minTemp)
                minTemp=val.get();
        if(val.get()>(int)maxTemp)
                maxTemp=val.get();
      }
       context.write(new Text(minTemp+"\t"+maxTemp), result);
    }
  }

  public static class normalizedMapper extends Mapper<Object, Text, Text, IntWritable>{
          public void map(Object key, Text value, Context context
                      ) throws IOException, InterruptedException {
            
                StringTokenizer itr = new StringTokenizer(value.toString());
                
                String headStr=itr.nextToken().toString();
                
                float headFloat=Float.parseFloat(headStr);
                
        headFloat=(headFloat-min)/(max-min);
        String headReplice= headFloat+" "; 
        String valueStr=value.toString(); 
        String changeValue=valueStr.replaceFirst(headStr, headReplice);
                context.write(new Text(changeValue), new IntWritable(1));        
          }
          
          @Override
                protected void setup(
                                Mapper<Object, Text, Text, IntWritable>.Context context)
                                throws IOException, InterruptedException {
                  
                          readDataFromHdfs rdfh=new readDataFromHdfs();
                          String minMaxStr=rdfh.getMinMaxStr();
                          String [] strs=minMaxStr.split("\t");
                    min=Float.parseFloat(strs[0]);
                    max=Float.parseFloat(strs[1]);
                        super.setup(context);
                }
  }
  
  public static void main(String[] args) throws Exception {
    Configuration conf = new Configuration();
    Job job = Job.getInstance(conf, "findMinMax");
    job.setJarByClass(normalizedVersion1.class);
    job.setMapperClass(findMaxMinMapper.class);
    job.setReducerClass(findMaxMinReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job, new Path(args[0]));
    FileOutputFormat.setOutputPath(job, new Path(args[1]));
    if(!job.waitForCompletion(true))
            System.exit(1);
    
    Job job1 = Job.getInstance(conf, "normalize");
    job1.setJarByClass(normalizedVersion1.class);
    job1.setMapperClass(normalizedMapper.class);
    job1.setOutputKeyClass(Text.class);
    job1.setOutputValueClass(IntWritable.class);
    FileInputFormat.addInputPath(job1, new Path(args[0]));
    FileOutputFormat.setOutputPath(job1, new Path(args[2]));
    System.exit(job1.waitForCompletion(true) ? 0 : 1);
    
  }
}
复制代码