hadoop的第一个程序WordCount

    xiaoxiao2021-03-25  100

    伪代码分析:

     

    执行步骤:

     1. map任务处理

    1.1读取输入文件内容,解析成key、value对。对输入文件的每一行,解析成key、value对。每一个键值对调用一次map函数。

    1.2写自己的逻辑,对输入的key、value处理,转换成新的key、value输出。

    1.3对输出的key、value进行分区。

    1.4对不同分区的数据,按照key进行排序、分组。相同key的value放到一个集合中。

    1.5(可选)分组后的数据进行归约。

    2.reduce任务处理

    2.1对多个map任务的输出,按照不同的分区,通过网络copy到不同的reduce节点。

    2.2对多个map任务的输出进行合并、排序。写reduce函数自己的逻辑,对输入的key、value处理,转换成新的key、value输出。

    2.3把reduce的输出保存到文件中。

    例子:实现WordCount

    WCMapper继承Mapper

    import java.io.IOException; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; //4个泛型中,前两个是指定mapper输入数据的类型,KEYIN是输入的key的类型,VALUEIN是输入的value的类型 //map 和 reduce 的数据输入输出都是以 key-value对的形式封装的 //默认情况下,框架传递给我们的mapper的输入数据中,key是要处理的文本中一行的起始偏移量,这一行的内容作为value public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable>{ //mapreduce框架每读一行数据就调用一次该方法 @Override protected void map(LongWritable key, Text value,Context context) throws IOException, InterruptedException { //具体业务逻辑就写在这个方法体中,而且我们业务要处理的数据已经被框架传递进来,在方法的参数中 key-value //key 是这一行数据的起始偏移量 value 是这一行的文本内容 //将这一行的内容转换成string类型 String line = value.toString(); //对这一行的文本按特定分隔符切分 String[] words = StringUtils.split(line, " "); //遍历这个单词数组输出为kv形式 k:单词 v : 1 for(String word : words){ context.write(new Text(word), new LongWritable(1)); } } }

    WCReducer继承Reducer

    import java.io.IOException; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable>{ //框架在map处理完成之后,将所有kv对缓存起来,进行分组,然后传递一个组<key,valus{}>,调用一次reduce方法 //<hello,{1,1,1,1,1,1.....}> @Override protected void reduce(Text key, Iterable<LongWritable> values,Context context) throws IOException, InterruptedException { long count = 0; //遍历value的list,进行累加求和 for(LongWritable value:values){ count += value.get(); } //输出这一个单词的统计结果 context.write(key, new LongWritable(count)); } }

    WordCount主函数类:

    import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; /** * 用来描述一个特定的作业 * 比如,该作业使用哪个类作为逻辑处理中的map,哪个作为reduce * 还可以指定该作业要处理的数据所在的路径 * 还可以指定改作业输出的结果放到哪个路径 * .... */ public class WCRunner { public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job wcjob = Job.getInstance(conf); //设置整个job所用的那些类在哪个jar包 wcjob.setJarByClass(WCRunner.class); //本job使用的mapper和reducer的类 wcjob.setMapperClass(WCMapper.class); wcjob.setReducerClass(WCReducer.class); //指定reduce的输出数据kv类型 wcjob.setOutputKeyClass(Text.class); wcjob.setOutputValueClass(LongWritable.class); //指定mapper的输出数据kv类型 wcjob.setMapOutputKeyClass(Text.class); wcjob.setMapOutputValueClass(LongWritable.class); //指定要处理的输入数据存放路径 FileInputFormat.setInputPaths(wcjob, new Path("hdfs://weekend110:9000/wc/srcdata/")); //指定处理结果的输出数据存放路径 FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://weekend110:9000/wc/output3/")); //将job提交给集群运行 wcjob.waitForCompletion(true); } }

    执行分两种情况:

     

     

    第一种:

    1.将程序打包成Runnerable jar file,并指定函数执行入口

     

    2.上传至linux

    3.用java -jar jar路径

    第二种:

    1.将程序打包成 jar file,并指定函数执行入口

    项目,右键——》export--》jar file (设置保存路径)

    2.上传至linux

    3.hadoop jar <jar在linux的路径> <main方法所在的类的全类名>

    注:启动hadoop集群,让hadoop分发jar。

    注意:

    1.编程时,注意导包错误。

     

     

     

    转载请注明原文地址: https://ju.6miu.com/read-16656.html

    最新回复(0)