编写mapreduce程序实例

您所在的位置：网站首页 › hadoop对两个文件中的数据进行合并和去重 › 编写mapreduce程序实例

编写mapreduce程序实例

2023-09-21 21:38| 来源: 网络整理| 查看: 265

每一行为一个日期及一个人名字，分为3个文件，其中存在多行重复数据，需要进行去重。

输入文件：

file1： 2006-6-9 a 2006-6-11 b 2006-6-21 c 2006-6-10 a 2006-6-11 b 2006-6-30 e 2006-6-9 a 2006-6-26 d

file2: 2006-6-9 a 2006-6-11 b 2006-6-9 a 2006-6-21 c 2006-6-11 b 2006-6-9 a 2006-6-30 e

file3: 2006-6-9 a 2006-6-11 b 2006-6-9 a 2006-6-21 c 2006-6-11 b 2006-6-26 d

期望输出： 2006-6-9 a 2006-6-10 a 2006-6-11 b 2006-6-21 c 2006-6-26 d 2006-6-30 e

这里写图片描述

编写代码：

package zmy.examples; import java.io.IOException; import java.util.StringTokenizer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; 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.Mapper.Context; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; import org.apache.hadoop.util.GenericOptionsParser; public class ZMY_DataDuplicate_dedup { //定义log变量 public static final Log LOG = LogFactory.getLog(FileInputFormat.class); public static class TokenizerMapper extends Mapper{ private static Text line=new Text();//每行数据 //实现map函数 public void map(Object key,Text value,Context context) throws IOException,InterruptedException{ System.out.println("map输入("+ key + "," + value + ")"); line=value; System.out.println("map输出("+ line + "," + new Text("") + ")"); context.write(line, new Text("")); } } public static class IntSumReducer extends Reducer{ //实现reduce函数 public void reduce(Text key,Iterable values,Context context) throws IOException,InterruptedException{ for (Text val : values) { System.out.println("reduce输入("+ key + "," + val.toString() + ")"); } System.out.println("reduce输出("+ key + "," + new Text("") + ")"); context.write(key, new Text("")); } } 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: data duplicate "); System.exit(2); } Job job = new Job(conf, "data duplicate"); job.setJarByClass(ZMY_DataDuplicate_dedup.class); job.setMapperClass(TokenizerMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); FileInputFormat.addInputPath(job, new Path(otherArgs[0])); FileOutputFormat.setOutputPath(job, new Path(otherArgs[1])); System.exit(job.waitForCompletion(true) ? 0 : 1); } }

运行过程： 1、单个文件的map输入输出与reduce输入输出过程：

file1

这里写图片描述