第一个wordcount程序

逃离我推掉我的手 2022-05-17 11:17 236阅读 0赞

*  在hadoop集群中，系统自带了许多mapreduce的程序例子，在\{%HOME%\}/hadoop-mapreduce/hadoop-mapreduce-examples.jar里面有许多小例子，可以在hadoop集群中直接运行命令.
 *  当然运行前需要在HDFS上创建input目录和output目录，
 *  input目录需要一个输入文件，比如/s\_tmp/input/tmp.txt
 *  接下来需要创建output目录，注意，output目录必须是不能存在的一个目录，否则，会报错。
 *  然后直接输入下面命令就可以运行自己的第一个mapreduce程序了。

hadoop jar hadoop-mapreduce-examples.jar wordcount /s_tmp/input/tmp.txt /s_tmp/output/

下面是我自己打包生成的mapreduce程序，在用hadoop jar命令执行的时候老是报错，提示找不到类，，，

package cn.sun.mapreduce;
    
    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;
    
    import java.io.IOException;
    import java.util.Iterator;
    import java.util.StringTokenizer;
    
    public class WordCount {
        public WordCount() {
        }
    
        public static void main(String[] args) throws Exception {
            Configuration conf = new Configuration();
            String[] otherArgs = (new GenericOptionsParser(conf, args)).getRemainingArgs();
            if (otherArgs.length < 2) {
                System.err.println("Usage: wordcount <in> [<in>...] <out>");
                System.exit(2);
            }
    
            Job job = Job.getInstance(conf, "word count");
            job.setJarByClass(WordCount.class);
            job.setMapperClass(WordCount.TokenizerMapper.class);
            job.setCombinerClass(WordCount.IntSumReducer.class);
            job.setReducerClass(WordCount.IntSumReducer.class);
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
    
            for(int i = 0; i < otherArgs.length - 1; ++i) { 
                FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
            }
    
            FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1]));
            System.exit(job.waitForCompletion(true) ? 0 : 1);
        }
    
        public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
            private IntWritable result = new IntWritable();
    
            public IntSumReducer() {
            }
    
            public void reduce(Text key, Iterable<IntWritable> values, Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
                int sum = 0;
    
                IntWritable val;
                for(Iterator i$ = values.iterator(); i$.hasNext(); sum += val.get()) { 
                    val = (IntWritable)i$.next();
                }
    
                this.result.set(sum);
                context.write(key, this.result);
            }
        }
    
        public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
            private static final IntWritable one = new IntWritable(1);
            private Text word = new Text();
    
            public TokenizerMapper() {
            }
    
            public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context) throws IOException, InterruptedException {
                StringTokenizer itr = new StringTokenizer(value.toString());
    
                while(itr.hasMoreTokens()) {
                    this.word.set(itr.nextToken());
                    context.write(this.word, one);
                }
    
            }
        }
    }

*  将这个程序打包生成jar包上传到服务器，然后运行命令

hadoop jar ~/BigData.jar cn.sun.mapreduce.WordCount /s_tmp/input/tmp.txt /s_tmp/output2/

*  对这个命令很不理解，然后上网搜了一下，大概意思是说，其中，/home/lina/example.jar是jar包所在目录，cn.sun.mapreduce.WordCount是因为main程序是放在cn.sun.mapreduce包下的WordCount.java下，/s\_tmp/input/tmp.txt是dfs文件系统下的tmp.txt输入文件的目录，/s\_tmp/output2/是设定的输出目录。

### mapreduce程序详解 ###

*  map类  
    Mapper 类，有四个泛型，分别是KEYIN、VALUEIN、KEYOUT、VALUEOUT，前面两个KEYIN、VALUEIN 指的是map 函数输入的参数key、value 的类型；后面两个KEYOUT、VALUEOUT 指的是map 函数输出的key、value 的类型；
    
    从代码中可以看出，在Mapper类和Reducer类中都使用了Hadoop自带的基本数据类型，例如String对应Text，long对应LongWritable，int对应IntWritable。这是因为HDFS涉及到序列化的问题，Hadoop的基本数据类型都实现了一个Writable接口，而实现了这个接口的类型都支持序列化。
 *  reduce类  
    Reducer 类，也有四个泛型，同理，分别指的是reduce 函数输入的key、value类型（这里输入的key、value类型通常和map的输出key、value类型保持一致）和输出的key、value 类型。  
    这里的reduce函数主要是将传入的（k2,v2）进行最后的合并统计，形成最后的统计结果。
 *  设置Main函数  
    在Main函数中，主要做了三件事：  
    1.一是指定输入、输出目录；  
    2.二是指定自定义的Mapper类和Reducer类；  
    3.三是提交作业；

// 1.0:指定输入目录
            FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));
            // 1.1:指定对输入数据进行格式化处理的类（可以省略）
            job.setInputFormatClass(TextInputFormat.class);
            // 1.2:指定自定义的Mapper类
            job.setMapperClass(MyMapper.class);
            // 1.3:指定map输出的<K,V>类型（如果<k3,v3>的类型与<k2,v2>的类型一致则可以省略）
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);
            // 1.4:分区（可以省略）
            job.setPartitionerClass(HashPartitioner.class);
            // 1.5:设置要运行的Reducer的数量（可以省略）
            job.setNumReduceTasks(1);
            // 1.6:指定自定义的Reducer类
            job.setReducerClass(MyReducer.class);
            // 1.7:指定reduce输出的<K,V>类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(LongWritable.class);
            // 1.8:指定输出目录
            FileOutputFormat.setOutputPath(job, new Path(OUTPUT_PATH));
            // 1.9:指定对输出数据进行格式化处理的类（可以省略）
            job.setOutputFormatClass(TextOutputFormat.class);
            // 2.0:提交作业
            boolean success = job.waitForCompletion(true);

### 高效编写mapreduce，了解一下ToolRunner与MNUnit！！！ ###