MapReduce编程规范（四）

arduinoCopy codepublic class SentimentAnalysis {

   public static class SentimentAnalysisMapper extends Mapper<Object, Text, Text, IntWritable> {

      private Map<String, Integer> dictionary = new HashMap<>();

      public void setup(Context context) throws IOException, InterruptedException {
         // 将词典文件读取到Map中
         URI[] cacheFiles = context.getCacheFiles();
         if (cacheFiles != null && cacheFiles.length > 0) {
            BufferedReader reader = new BufferedReader(new FileReader(new File(cacheFiles[0].getPath())));
            String line;
            while ((line = reader.readLine()) != null) {
               String[] tokens = line.split("\t");
               String word = tokens[0];
               int score = Integer.parseInt(tokens[1]);
               dictionary.put(word, score);
            }
            reader.close();
         }
      }

      public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
         int sentimentScore = 0;
         String[] words = value.toString().split(" ");
         for (String w : words) {
            Integer score = dictionary.get(w);
            if (score != null) {
               sentimentScore += score;
            }
         }
         context.write(new Text("sentiment score"), new IntWritable(sentimentScore));
      }
   }

   public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
      Job job = Job.getInstance(conf, "sentiment analysis");
      job.setJarByClass(SentimentAnalysis.class);
      job.setMapperClass(SentimentAnalysisMapper.class);
      job.addCacheFile(new URI(args[1])); // 添加词典文件到分布式缓存
      job.setOutputKeyClass(Text.class);
      job.setOutputValueClass(IntWritable.class);
      FileInputFormat.addInputPath(job, new Path(args[0]));
      FileOutputFormat.setOutputPath(job, new Path(args[2]));
      System.exit(job.waitForCompletion(true) ? 0 : 1);
   }
}