需要配置Windows下的hadoop环境,大家可以去参考我的另一篇文章
Windows下HDFS的环境准备——HDFS相关的客户端操作_你可以自己看的博客-CSDN博客
创建Maven工程
设置Maven配置
导包
org.apache.hadoop hadoop-client3.1.3 junit junit4.12 test org.slf4j slf4j-log4j121.7.30
添加日志配置文件log4j.properties
log4j.rootLogger=INFO, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n log4j.appender.logfile=org.apache.log4j.FileAppender log4j.appender.logfile.File=target/spring.log log4j.appender.logfile.layout=org.apache.log4j.PatternLayout log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n
编写Mapper类
package com.gk.mapreduce.wordcount; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class WordCountMapper extends Mapper{ private Text outK = new Text(); private IntWritable outV = new IntWritable(1); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 1 获取一行 String line = value.toString(); // 2 切割 String[] words = line.split(" "); // 3 输出 for (String word : words) { outK.set(word); context.write(outK, outV); } } }
编写Reduce类
package com.gk.mapreduce.wordcount; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class WordCountReducer extends Reducer{ private IntWritable value = new IntWritable(); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int sum = 0; // 遍历读取值 for (IntWritable value : values) { // 1.累加求和 sum += value.get(); } // 2.输出 value.set(sum); context.write(key,value); } }
创建Driver类
package com.gk.mapreduce.wordcount;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.IOException;
public class WordCountDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
// 1.获取配置信息以及获取job对象
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2.关联Driver程序的jar包
job.setJarByClass(WordCountDriver.class);
// 3.关联Mapper和reducer的Jar
job.setMapperClass(WordCountMapper.class);
job.setReducerClass(WordCountReducer.class);
// 4.设置Mapper输出的k,y类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(IntWritable.class);
// 5.设置最终输出k,v类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
// 6.设置输入输出路径
FileInputFormat.setInputPaths(job, new Path("E:\SGG-Hadoop\wcInput"));
FileOutputFormat.setOutputPath(job, new Path("E:\SGG-Hadoop\wcOutput"));
// 7.提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
然后执行WordCountDriver下的main方法
显示已经执行成功了
现在我们去看一下输出文件
已经成功统计出输入文件中单词的次数了
现在我们把这个代码打包上传到集群去试一下,在这之前我们先去修改一些东西
首先,在pom.xml文件下添加如下代码,当然,其实这些代码可以不加,含义是打包的时候将依赖包一并打入进jar包中,我们的hadoop集群中是有相关jar包的,所以可以不需要
maven-compiler-plugin 3.6.1 1.8 1.8 maven-assembly-plugin jar-with-dependencies make-assembly package single
然后复制一份一模一样的代码出来,但是文件夹变了
然后修改WordCountDriver下的代码,将文件输入输出路径改为动态的
然后进行打包
现在我们有两个jar包了,分别是带依赖和不带依赖的
我们上传不带依赖的,也就是9KB的那个jar包到我们的集群中
然后我们使用这个jar包测试mapreduce,执行以下命令(需要保证有输入文件,具体可以看我另外一篇文章Hadoop单机和完全分布式自带Mapreduce测试_你可以自己看的博客-CSDN博客)
hadoop jar MapreduceDemo-1.0-SNAPSHOT.jar com.gk.mapreduce.wordcount2.WordCountDriver /wcinput /wcoutput
可以发现这样是用我们的自己写的代码执行的mapreduce测试。



