大数据实战——基于Hadoop的Mapreduce编程实践案例的设计与实现

import java.io.IOException;
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 org.apache.hadoop.util.GenericOptionsParser;
public class spark {
    public static class Map extends Mapper<Object, Text, IntWritable, IntWritable> {
        private static IntWritable data = new IntWritable();
        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            data.set(Integer.parseInt(line));
            context.write(data, new IntWritable(1));
        }
    }
    public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable> {
        private static IntWritable lineNum = new IntWritable(1);
        @Override
        public void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            for (IntWritable val : values) {
                context.write(lineNum, key);
                lineNum = new IntWritable(lineNum.get() + 1);}}}
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // 这句话很关键
        // conf.set("mapred.job.tracker", "192.168.1.2:9001");
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length != 2) {
            System.err.println("Usage: Data Sort <in> <out>");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "spark");
        job.setJarByClass(spark.class);
        // 设置 Map 和 Reduce 处理类
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        // 设置输出类型
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        // 设置输入和输出目录
        FileInputFormat.setInputPaths(job, new Path("hdfs://localhost:9000/spark/dataFile6.txt"));
        FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/spark/output1"));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

package Mapreduce;

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.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 org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;

public class scoreSort {
    public static class scoreSortMapper extends Mapper<LongWritable, Text, Text, DoubleWritable> {
        private final Text k2 = new Text();
        private final DoubleWritable v2 = new DoubleWritable();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 通过空格进行分割
            StringTokenizer tokenizer = new StringTokenizer(value.toString());
            // 迭代拿出
            while (tokenizer.hasMoreTokens()) {
                k2.set(tokenizer.nextToken());
                v2.set(Double.parseDouble(tokenizer.nextToken()));
                context.write(k2, v2);
            }
        }
    }

    public static class scoreSortReduce extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
        @Override
        protected void reduce(Text key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {
            double sum = 0;
            int count = 0;
            for (DoubleWritable val : values) {
                sum += val.get();
                count++;
            }
            context.write(key, new DoubleWritable(sum / count));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        /* 配置文件 */
        Configuration config = new Configuration();
        Job job = Job.getInstance(config, "App");
        /* mapper操作 */
        job.setMapperClass(scoreSortMapper.class);
        /* 设置map后输出数据类型，如果不设置会默认输出<IntWritable,Text>类型，会报错 */
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(DoubleWritable.class);
        /* 设置输出文件类型 */
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);
        /* combiner操作 */
        job.setCombinerClass(scoreSortReduce.class);
        /* reduce操作 */
        job.setReducerClass(scoreSortReduce.class);
        /* 设置输入、输出目录，输出目录不能存在 */
        /* 设置输入输出的目录 */
        Path inputpath = new Path("hdfs://localhost:9000/user/hadoop/data/cai.txt");
        Path outpath = new Path("hdfs://localhost:9000/out4");
        /* 设置需要计算的文件 */
        FileInputFormat.addInputPath(job, inputpath);
        /* 删除多余的目录 */
        //MpUtil.delOutPut(config, outpath);
        FileOutputFormat.setOutputPath(job, outpath);
        /* 0表示正常退出，1表示错误退出 */
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

package cbcPR;

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.Text;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
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;

public class PageRank {

  public static enum PageCouter{  
    TotalPage,  
    ConvergePage  
    };
    /*map过程*/
  public static class TokenizerMapper extends Mapper<Object, Text, Text, Text>{
      public void map(Object key, Text value, Context context
                     ) throws IOException, InterruptedException {
          StringTokenizer itr = new StringTokenizer(value.toString());
          String id = itr.nextToken(); //当前网页
          float weight = Float.parseFloat(itr.nextToken()); //当前网页的权值
          float out_weight = weight / itr.countTokens(); //为每个网页投票的权值
          String outs = "";
          while (itr.hasMoreTokens()) {
              String out_id = itr.nextToken();
              outs += out_id + " ";
              context.write(new Text(out_id), new Text("&" + out_weight));
          }
          context.write(new Text(id), new Text(outs)); 
          context.write(new Text(id), new Text("#" + weight)); 
          //'#'表示当前的权值，用于reduce阶段判断是否收敛
          //枚举计数器，计算有几个Page
          context.getCounter(PageCouter.TotalPage).increment(1);
      }
  }

    /*reduce过程*/
  public static class WeightReducer extends Reducer<Text,Text,Text,Text> {
      public void reduce(Text key, Iterable<Text> values, Context context
                        ) throws IOException, InterruptedException {
          float sum_weight = 0; //新的权值
          float pre_weight = 0; //上一轮的权值，比较是否收敛
          String outs = "";
          for (Text val : values) {
              String tmp = val.toString();
              if (tmp.startsWith("&")) 
                  sum_weight+=Float.parseFloat(tmp.substring(1));
              else if (tmp.startsWith("#"))
                  pre_weight =Float.parseFloat(tmp.substring(1)); 
              else outs = tmp;
          }
          sum_weight = 0.8f * sum_weight + 0.2f * 0.125f; //平滑处理，处理终止点和陷阱 
          context.write(key, new Text(sum_weight + " " + outs));
          if (sum_weight == pre_weight)        //枚举计数器，计算有几个已收敛
              context.getCounter(PageCouter.ConvergePage).increment(1);
      }
  }

  public static void main(String[] args) throws Exception {
      Configuration conf = new Configuration();
      String in_path = args[0];
      String out_path = args[1] + "/iter";
      for (int i = 0; i < 50; i++) { //设置最多只迭代50次，也可用while
          Job job = Job.getInstance(conf, "page rank");
          job.setJarByClass(PageRank.class);
          job.setMapperClass(TokenizerMapper.class);
          job.setReducerClass(WeightReducer.class);
          job.setOutputKeyClass(Text.class);
          job.setOutputValueClass(Text.class);
          FileInputFormat.addInputPath(job, new Path(in_path));
          FileOutputFormat.setOutputPath(job, new Path(out_path + i));
          job.setNumReduceTasks(2); //设置reduce数量为2，模拟大文件
          in_path = out_path + i;   //设置下一轮的输入
          job.waitForCompletion(true);

          Counters counters = job.getCounters(); 
          Counter counter = counters.findCounter(PageCouter.TotalPage);
          long totalpage=counter.getValue();// 获取计数值
          counter = counters.findCounter(PageCouter.ConvergePage);
          long convergepage = counter.getValue(); //收敛计数值
          System.out.println("total page: " + totalpage);
          System.out.println("converge page: " + convergepage);
          if (totalpage == convergepage) {
              System.out.print("converge at iteration: " + i);
              break;
          }
      }
  }
}