大数据系列之分布式计算批处理引擎MapReduce实践

?WordCount:  1.应用场景,在大量文件中存储了单词,单词之间用空格分隔  2.类似场景:搜索引擎中,统计最流行的N个搜索词,统计搜索词频率,帮助优化搜索词提示。  3.采用MapReduce执行过程如图       3.1MapReduce将作业的整个运行过程分为两个阶段        3.1.1Map阶段和Reduce阶段            Map阶段由一定数量的Map Task组成            输入数据格式解析:InputFormat            输入数据处理:Mapper            数据分组:Partitioner        3.1.2Reduce阶段由一定数量的Reduce Task组成            数据远程拷贝            数据按照key排序            数据处理:Reducer            数据输出格式:OutputFormat?  4.介绍代码结构  4.1 pom.xml


    4.0.0

    hadoop
    hadoop.mapreduce
    1.0-SNAPSHOT

    
        
            aliyun
            http://maven.aliyun.com/nexus/content/groups/public/
        
    
    
        
            org.apache.hadoop
            hadoop-yarn-client
            2.7.3
        
        
            org.apache.hadoop
            hadoop-common
            2.7.3
        
        
            org.apache.hadoop
            hadoop-mapreduce-client-jobclient
            2.7.3
        
    

    
        
            
                maven-assembly-plugin
                2.3
                
                    dist
                    true
                    
                        jar-with-dependencies
                    
                
                
                    
                        make-assembly
                        package
                        
                            single
                        
                    
                
            
        
    


?  4.2 WordCount.java
package hadoop.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;

public class WordCount {

    public static class WordCountMap
            extends Mapper {

        public void map(Object key,Text value, Context context) throws IOException, InterruptedException {
            //在此处写map代码
            String[] lines = value.toString().split(" ");
            for (String word : lines) {
                context.write(new Text(word), new IntWritable(1));
            }
        }
    }

    public static class WordCountReducer
            extends Reducer {

        public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
            //在此处写reduce代码
            int count=0;
            for (IntWritable cn : values) {
                count=count+cn.get();
            }
            context.write(key, new IntWritable(count));
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: wordcount  [...] ");
            System.exit(2);
        }
        Job job = Job.getInstance(conf, "word count");
        job.setJarByClass(WordCount.class);
        //设置输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //设置输出路径
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        //设置实现map函数的类
        job.setMapperClass(WordCountMap.class);
        //设置实现reduce函数的类
        job.setReducerClass(WordCountReducer.class);

        //设置map阶段产生的key和value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        //设置reduce阶段产生的key和value的类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //提交job
        job.waitForCompletion(true);

        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);
    }

}
  4.3 data目录下文件内容:    to.txt?
hadoop spark hive hbase hive
?   t1.txt
hive spark mapReduce spark
?   ?t2.txt
sqoop spark hadoop
??5. 数据准备  5.1?maven 打jar包为hadoop.mapreduce-1.0-SNAPSHOT.jar,传入master服务器上   ?  5.2 将需要计算的数据文件放入datajar/in (临时目录无所谓在哪里)     5.3 启动hadoop ,关于hadoop安装可参考我写的文章?    将datajar/in文件传至hdfs 上
hadoop fs -put in /in  
#查看文件
hadoop fs -ls -R /in
 5.4 执行jar  两种命令方式
#第一种:hadoop jar
hadoop jar hadoop.mapreduce-1.0-SNAPSHOT.jar hadoop.mapreduce.WordCount /in/* /out

#OR 
#第二种:yarn jar
yarn jar hadoop.mapreduce-1.0-SNAPSHOT.jar hadoop.mapreduce.WordCount /in/* /yarnOut
?  5.5.执行后输出内容分别如图hadoop jar ...结果yarn jar ... 结果? 6.查看结果内容
#查看hadoop ja 执行后输出结果目录
hadoop fs -ls -R /out

#查看yarn jar 执行后输出结果目录
hadoop fs -ls -R /yarnOut
?  目录说明:目录中_SUCCESS 是日志文件,part-r-00000是计算结果文件  查看计算结果
#查看out/part-r-00000文件
 hadoop fs -text /out/part-r-00000

#查看yarnOut/part-r-00000文件
 hadoop fs -text /yarnOut/part-r-00000
??完~~~,Java代码内容已上传至GitHub?https://github.com/fzmeng/MapReduceDemo?

相关内容推荐