MapReduce明星搜索指数统计，找出人气王详解大数据

package com.buaa; 
 
import java.io.IOException; 
 
import org.apache.hadoop.conf.Configuration; 
import org.apache.hadoop.conf.Configured; 
import org.apache.hadoop.fs.FileSystem; 
import org.apache.hadoop.fs.Path; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Job; 
import org.apache.hadoop.mapreduce.Mapper; 
import org.apache.hadoop.mapreduce.Partitioner; 
import org.apache.hadoop.mapreduce.Reducer; 
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; 
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; 
import org.apache.hadoop.util.Tool; 
import org.apache.hadoop.util.ToolRunner; 
 
/**  * @ProjectName CountStarSearchIndex 
* @PackageName com.buaa 
* @ClassName SearchStarIndex 
* @Description 统计分别统计出男女明星最大搜索指数 
* @Author 刘吉超 
* @Date 2016-05-12 16:30:23 
*/ public class SearchStarIndex extends Configured implements Tool { // 分隔符/t private static String TAB_SEPARATOR = "/t"; // 男 private static String MALE = "male"; // 女 private static String FEMALE = "female"; 
     /* 
     * 解析明星数据 
     */ public static class IndexMapper extends Mapper<Object, Text, Text, Text> { /* 
         * 每次调用map(LongWritable key, Text value, Context context)解析一行数据。 
         * 每行数据存储在value参数值中。然后根据'/t'分隔符，解析出明星姓名，性别和搜索指数 
         */ public void map(Object key, Text value, Context context) throws IOException, InterruptedException { // 将数据解析为数组 
            String[] tokens = value.toString().split(TAB_SEPARATOR); 
             if(tokens != null && tokens.length >= 3){ // 性别 
                String gender = tokens[1].trim(); // 名称、关注指数 
                String nameHotIndex = tokens[0].trim() + TAB_SEPARATOR + tokens[2].trim(); 
                 // 输出key=gender value=name+hotIndex 
                context.write(new Text(gender), new Text(nameHotIndex)); 
            } 
        } 
    } 
     /* 
     * 根据性别对数据进行分区，将 Mapper的输出结果均匀分布在 reduce上 
     */ public static class IndexPartitioner extends Partitioner<Text, Text> {          
        @Override public int getPartition(Text key, Text value, int numReduceTasks) {  // 按性别分区 
            String sex = key.toString(); 
             // 默认指定分区 0 if(numReduceTasks == 0) return 0; 
             // 性别为男，选择分区0 if(MALE.equals(sex)){            return 0; 
            }else if(FEMALE.equals(sex)){ // 性别为女，选择分区1 return 1 % numReduceTasks; 
            }else // 性别未知，选择分区2 return 2 % numReduceTasks; 
            
        } 
    } 
     /* 
     * 定义Combiner，对 map端的输出结果，先进行一次合并，减少数据的网络输出 
     */ public static class IndexCombiner extends Reducer<Text, Text, Text, Text> { 
         
        @Override public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException { int maxHotIndex = Integer.MIN_VALUE; 
            String name= ""; 
             for (Text val : values) { 
                String[] valTokens = val.toString().split(TAB_SEPARATOR); 
                 int hotIndex = Integer.parseInt(valTokens[1]); 
                 if(hotIndex > maxHotIndex){ 
                    name = valTokens[0]; 
                    maxHotIndex = hotIndex; 
                } 
            } 
             
            context.write(key, new Text(name + TAB_SEPARATOR + maxHotIndex)); 
        } 
    } 
     /* 
     * 统计男、女明星最高搜索指数 
     */ public static class IndexReducer extends Reducer<Text, Text, Text, Text> { /* 
         * 调用reduce(key, Iterable< Text> values, context)方法来处理每个key和values的集合。 
         * 我们在values集合中，计算出明星的最大搜索指数 
         */ 
        @Override public void reduce(Text key, Iterable<Text> values, Context context)throws IOException, InterruptedException { int maxHotIndex = Integer.MIN_VALUE; 
            String name = " "; 
             // 根据key，迭代 values集合，求出最高搜索指数 for (Text val : values) { 
                String[] valTokens = val.toString().split(TAB_SEPARATOR); 
                 int hotIndex = Integer.parseInt(valTokens[1]); 
                 if (hotIndex > maxHotIndex) { 
                    name = valTokens[0]; 
                    maxHotIndex = hotIndex; 
                } 
            } 
             
            context.write(new Text(name), new Text(key + TAB_SEPARATOR + maxHotIndex)); 
        } 
    } 
     
    @SuppressWarnings("deprecation") 
    @Override public int run(String[] args) throws Exception { // 读取配置文件 
        Configuration conf = new Configuration(); 
         // 如果目标文件夹存在，则删除 
        Path mypath = new Path(args[1]); 
        FileSystem hdfs = mypath.getFileSystem(conf); if (hdfs.isDirectory(mypath)) { 
            hdfs.delete(mypath, true); 
        } 
 // 新建一个任务 
        Job job = new Job(conf, "searchStarIndex"); // 主类 
        job.setJarByClass(SearchStarIndex.class); 
         // reduce的个数设置为2 
        job.setNumReduceTasks(2); // 设置Partitioner类 
        job.setPartitionerClass(IndexPartitioner.class); 
         // Mapper 
        job.setMapperClass(IndexMapper.class); // Reducer 
        job.setReducerClass(IndexReducer.class); 
         // map 输出key类型 
        job.setMapOutputKeyClass(Text.class); // map 输出value类型 
        job.setMapOutputValueClass(Text.class); 
         // 设置Combiner类 
        job.setCombinerClass(IndexCombiner.class); 
         // 输出结果 key类型 
        job.setOutputKeyClass(Text.class); // 输出结果 value类型 
        job.setOutputValueClass(Text.class); 
         // 输入路径 
        FileInputFormat.addInputPath(job, new Path(args[0])); // 输出路径 
        FileOutputFormat.setOutputPath(job, new Path(args[1])); 
         // 提交任务 return job.waitForCompletion(true) ? 0 : 1; 
    } 
     public static void main(String[] args) throws Exception { 
        String[] args0 = {  "hdfs://ljc:9000/buaa/index/index.txt", "hdfs://ljc:9000/buaa/index/out/" 
        }; int ec = ToolRunner.run(new Configuration(), new SearchStarIndex(), args0); 
        System.exit(ec); 
    } 
}