学步园

边数据分布：

边数据是作业所需的额外的只读数据，以辅助处理主数据集。所面临的挑战是如何使所用Map和Reduce任务都能够方便而高效的使用边数据。

我们用DistributedCache Api实现：

1）刚开始用1.0.0的api怎么也不行，运行到patternsFiles = DistributedCache.getLocalCacheFiles(job);patternsFiles一直为空。google后说是老版本的好实现，就实验了下老版本：
 

import java.io.*;<br /> import java.util.*;</p> <p> import org.apache.hadoop.fs.Path;<br /> import org.apache.hadoop.filecache.DistributedCache;<br /> import org.apache.hadoop.conf.*;<br /> import org.apache.hadoop.io.*;<br /> import org.apache.hadoop.mapred.*;<br /> import org.apache.hadoop.util.*;</p> <p> public class testDistributeCache extends Configured implements Tool {<br /> public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, IntWritable> {</p> <p> public void configure(JobConf job) {<br /> Path[] patternsFiles;<br /> boolean test = job.getBoolean("wordcount.skip.patterns", false);<br /> try {<br /> <span style="color:#ff6666;">patternsFiles = DistributedCache.getLocalCacheFiles(job);<br /> </span> int test4=0;<br /> } catch (IOException ioe) {<br /> System.err.println("Caught exception while getting cached files: " + StringUtils.stringifyException(ioe));<br /> }<br /> }</p> <p> @Override<br /> public void map(LongWritable arg0, Text arg1,<br /> OutputCollector<Text, IntWritable> arg2, Reporter arg3)<br /> throws IOException {<br /> arg2.collect(arg1, new IntWritable(1));<br /> // TODO Auto-generated method stub</p> <p> }<br /> }</p> <p> public int run(String[] args) throws Exception {<br /> JobConf conf = new JobConf();</p> <p> conf.setOutputKeyClass(Text.class);<br /> conf.setOutputValueClass(LongWritable.class);</p> <p> conf.setMapperClass(Map.class);</p> <p> conf.setInputFormat(TextInputFormat.class);<br /> conf.setOutputFormat(TextOutputFormat.class);</p> <p> DistributedCache.addCacheFile(new Path("/testDisFile").toUri(), conf);<br /> conf.setBoolean("wordcount.skip.patterns", true);</p> <p> FileInputFormat.setInputPaths(conf,new Path("/testDisIn") );<br /> FileOutputFormat.setOutputPath(conf, new Path("/testDisOut"));</p> <p> JobClient.runJob(conf);<br /> return 0;<br /> }</p> <p> public static void main(String[] args) throws Exception {<br /> int res = ToolRunner.run(new Configuration(), new testDistributeCache(), args);<br /> System.exit(res);<br /> }<br /> }

测试

12/04/11 08:40:32 INFO filecache.TrackerDistributedCacheManager: Creating testDisFile in /tmp/hadoop-mjiang/mapred/local/archive/-3764210432234018452_-1245135191_482447420/localhost-work-9116784916847441351 with rwxr-xr-x

12/04/11 08:40:32 INFO filecache.TrackerDistributedCacheManager: Cached /testDisFile as /tmp/hadoop-mjiang/mapred/local/archive/-3764210432234018452_-1245135191_482447420/localhost/testDisFile
12/04/11 08:40:32 INFO filecache.TrackerDistributedCacheManager: Cached /testDisFile as /tmp/hadoop-mjiang/mapred/local/archive/-3764210432234018452_-1245135191_482447420/localhost/testDisFile

12/04/11 08:40:32

debug时

patternsFiles = DistributedCache.getLocalCacheFiles(job);<br /> 显示patternsFiles有值，说明是conf的问题

2）

－－－The Hadoop Map-Reduce framework spawns one map task for each
InputSplit generated by the
InputFormat for the job. Mapper implementations can access theConfiguration
for the job via the
～Context.getConfiguration().

The framework first calls
setup(org.apache.hadoop.mapreduce.Mapper.Context), followed by
map(Object, Object, Context) for each key/value pair in the
InputSplit. Finally
cleanup(Context) is called.

－－－

jobconf 或job或者configuratin只能有一个，或者是关联的。

－－－

又连续测试了很多情况最后终于OK了

import java.io.*;<br /> import java.net.URI;</p> <p>import org.apache.hadoop.conf.Configuration;<br /> import org.apache.hadoop.filecache.DistributedCache;<br /> import org.apache.hadoop.fs.Path;<br /> import org.apache.hadoop.io.*;<br /> import org.apache.hadoop.mapred.JobConf;<br /> import org.apache.hadoop.mapreduce.Job;<br /> import org.apache.hadoop.mapreduce.Mapper;<br /> import org.apache.hadoop.mapreduce.Reducer;<br /> import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;<br /> import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;<br /> import org.apache.hadoop.mapreduce.*;</p> <p> public class getBinaryFpAndPredict {<br /> //private BufferedReader modelBR = new BufferedReader(new<br /> FileReader("/home/mjiang/java/eclipse/hadoop/Target-1/data/models.txt"));</p> <p> public static class TargetMapper extends Mapper<LongWritable,<br /> Text, Text, Text> { </p> <p> private Path[] modelPath;<br /> private BufferedReader modelBR ;<br /> public void setup(Context context) throws<br /> IOException,InterruptedException{</p> <p> //Configuration conf = new Configuration(); /testS 为空<br /> Configuration conf = context.getConfiguration(); //testS 为空<br /> boolean test = conf.getBoolean("wordcount.skip.patterns",<br /> false);<br /> modelPath = DistributedCache.getLocalCacheFiles(conf);</p> <p> int testi = 0;<br /> modelBR = new BufferedReader(new<br /> FileReader(modelPath[0].toString()));<br /> } </p> <p> public void map(LongWritable key, Text value, Context context)<br /> throws IOException, InterruptedException {<br /> //获取model的文件<br /> String hexFp = value.toString().replaceAll("\t", "").replaceAll("<br /> ", ""); </p> <p> context.write(new Text("1"),new<br /> Text(hexFp+"~~~~~~~"+modelBR.readLine()));<br /> }<br /> }</p> <p> static class TargetReducer extends Reducer<Text, Text, Text,<br /> Text> {<br /> public void reduce(Text key, Text values, Context context) throws<br /> IOException, InterruptedException {<br /> context.write(key, values);<br /> }<br /> }</p> <p> public static void main(String[] args) throws Exception {</p> <p> Job job = new Job();<br /> Configuration conf = job.getConfiguration();<br /> conf.setBoolean("wordcount.skip.patterns", true);<br /> DistributedCache.addCacheFile(new<br /> Path("target/model/all.model").toUri(), conf);</p> <p> job.setJarByClass(getBinaryFpAndPredict.class);</p> <p> FileInputFormat.addInputPath(job, new<br /> Path("target/hexFp/all.hexFp"));<br /> FileOutputFormat.setOutputPath(job, new Path("/testOut"));</p> <p> job.setInputFormatClass(hexFpTextInputFormat.class);</p> <p> job.setMapperClass(TargetMapper.class);<br /> job.setReducerClass(TargetReducer.class);</p> <p> job.setOutputKeyClass(Text.class);<br /> job.setOutputValueClass(Text.class);</p> <p> boolean test = conf.getBoolean("wordcount.skip.patterns", false);<br /> //MapContext jobConf= = new MapContext(conf, null);<br /> Path[] modelPath = DistributedCache.getLocalCacheFiles(conf);</p> <p> System.exit(job.waitForCompletion(true) ? 0 : 1);<br /> }<br /> }在调试时还不行，运行就OK了。木有保存吗？？郁闷