学步园

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

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

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

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

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

闲话少说，上代码：

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

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