输入数据
test1 test2 test2 test3 test3 test3 test5 test5 test4 test5 test5 test4 test5 test4 test4
Maven必须配置
注意:Windos本地运行需要确定本地有Hadoop依赖并确保和Pom配置文件中版本一致,WordCountDriver中第6点输入输出需要自行修改
4.0.0 com.test2 mapredceDemo11.0-SNAPSHOT org.apache.hadoop hadoop-client3.0.1 junit junit4.12 org.slf4j slf4j-log4j121.7.30 maven-compiler-plugin 3.6.1 1.8 maven-assembly-plugin jar-with-dependencies make-assembly package single
resources目录下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类实现(WordCountMapper)
package com.atguigu.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{ // 定义一个Text对象,封装数据 private Text k = new Text(); // 定义一个IntWritable对象,直接封装1为值 private IntWritable v = 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) { k.set(word); context.write(k, v); } } }
自定义Reducer类实现(WordCountReducer)
package com.test.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{ // 定义变量统计词频 int sum; // 定义一个IntWritable对象,以便封装数据 IntWritable v = new IntWritable(); @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { // 1. 初始化词频 sum = 0; // 2. 统计词频 for (IntWritable count : values) sum += count.get(); // 3. 封装value v.set(sum); // 4. 输出 context.write(key, v); } }
自定义Driver类实现(WordCountDriver)
package com.atguigu.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.创建配置信息Configuration对象并获取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输出的kv类型 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(IntWritable.class); // 5. 设置最终输出的kv类型 job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); // 6.设置输入和输出路径 FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); // 7.提交job boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }
Linux集群环境下运行命令(代码打包后改名为wc.jar):
hadoop jar wc.jar com.atguigu.mapreduce.wordcount.WordCountDriver /input /output
输出数据
test1 1 test2 2 test3 3 test4 4 test5 5
欢迎分享,转载请注明来源:内存溢出
评论列表(0条)