MapReduce下的数据传递

好久没写点文章了，上一篇文章都是4月的了。实在不应该。实习是解决了，但随后的事也多了去了。实验报告、课程设计、比赛、论文、软考、考试。我嘞个去，杂那么多哟～～ 忙是忙，但还是不要忘里学习。今天来讨论上次做TF-IDF时遇到的问题---MapReduce下的数据传递。

我们写MapReduce程序时，有时需要将一定的值（这里是少量 的）从cilent传到map或者reduce.又或者从map传到reduce。

我们先来讨论比较简单的第一种。

解决办法，在main()函数中通过xml文件设定需要传送的值。然后在map函数中读取就行了。

第二种肯定不能使用第一种的方法。因为map阶段跟reduce阶段不一定在同一台机子上，就算map设定了值，reduce也不能够读出来。这里我们就只有使用笨办法了，直接从文件里读。

闲话少说，上代码：

public class xml_test {
	public static int getFileInt(String filename) //从文件中读取预设值
	{
		int temp = 0;
		Configuration config = new Configuration();
		FSDataInputStream dis = null;
		try {
				ByteArrayOutputStream baos = new ByteArrayOutputStream(); 
				FileSystem hdfs = FileSystem.get(config);
				dis = hdfs.open(new Path(filename));
				IOUtils.copyBytes(dis, baos, 4096, false); //写入ByteArrayOutputStream
				String str = baos.toString();  //这里我假定只有一行，多行可以使用循环。
				str = str.substring(0, str.length() - 1); //最后一个是回车，需要过滤掉，不然在整形转换时会出错
				temp = Integer.parseInt(str);
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}	
			finally{
				IOUtils.closeStream(dis);
			}
		return temp;
}
	public static  class wordcountMapper extends
		Mapper<LongWritable, Text, Text, IntWritable>{
		int temp2 = getFileInt("/user/jackydai/int"); //从文件中读取预设的值
		//这里也可以通过args参数传一个文件进来，这样更灵活
		private Text word = new Text();
		public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
			int temp1 = 0;
			Configuration mapconf = context.getConfiguration();
			temp1 = mapconf.getInt("count", 0); //map读取值
			IntWritable one = new IntWritable(temp1 + temp2); //求和后输出，检测是否正确
			String line = value.toString();
			StringTokenizer itr = new StringTokenizer(line);
			while(itr.hasMoreElements()){
				word.set(itr.nextToken());
				context.write(word, one);
			}
		}
	}
	public static  class wordcountReduce extends
		Reducer<Text, IntWritable, Text, IntWritable>{
		public void reduce(Text key, Iterable<IntWritable>values, Context context)throws IOException, InterruptedException{
			int sum = 0;
			for (IntWritable str : values){
				sum += str.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}
	public static  void main(String args[])throws Exception{
		
		Configuration conf = new Configuration();
		
		conf.setInt("count", 2); //设置值为2，需要注意的是设置值需要在new job之前
		
		Job job = new Job(conf, "xml_test");
		
		job.setJarByClass(xml_test.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		job.setMapperClass(wordcountMapper.class);
		job.setReducerClass(wordcountReduce.class);
		job.setCombinerClass(wordcountReduce.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[1]));
		FileOutputFormat.setOutputPath(job, new Path(args[2]));
		
		job.waitForCompletion(true);
	}
}

注释我基本写在了代码里。