使用DistributedCache出错

现象：在Hadoop 2.x上，用新的DistributedCache的API，在mapper中会获取不到这个cache文件。

将HDFS文件添加到distributed cache中：
        job.addCacheFile(new Path(inputFileOnHDFS).toUri());
在mapper的setup()方法中：
        Configuration conf = context.getConfiguration();
        URI[] localCacheFiles = context.getCacheFiles();
        readCacheFile(localCacheFiles[0]);
        其中，readCacheFile()是我们自己的读取cache文件的方法，可能是这样做的（仅举个例子）：
        private static void readCacheFile(URI cacheFileURI) throws IOException {
  BufferedReader reader = new BufferedReader(new FileReader(cacheFileURI.getPath()));
  String line;
  while ((line = reader.readLine()) != null) {
    //TODO: your code here
  }
  reader.close();
}
问题发现 context.getCacheFiles() 总是返回null，也就是你无法读到cache文件。

最好贴出代码，相信你的思路是没有错的。

这里提几个不一样的地方，你可以尝试修改：
    job.addCacheFile(new Path(inputFileOnHDFS).toUri());

job.addCacheFile(new URI(inputFileOnHDFS));
例如：
job.addCacheFile(new URI("/myapp/lookup.dat#lookup.dat"));


把这个
URI[] localCacheFiles = context.getCacheFiles();
改成下面其中一种形式：
private Path[]   localArchives = context.getLocalCacheArchives();
private Path[]  localFiles = context.getLocalCacheFiles();

howtodown 发表于 2014-9-27 00:42
最好贴出代码，相信你的思路是没有错的。

这里提几个不一样的地方，你可以尝试修改：

package hadoop.mapjoin;

import hadoop.other.Stock;
import hadoop.other.StockPrices;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MapSideJoin extends Configured implements Tool {
        enum TESTW{
                MISSING,
                HELLO;
        }
        public static class MapSideJoinMapper extends Mapper<LongWritable, Text, Stock, StockPrices> {
                private String stockSymbol;
                private HashMap<Stock,Double> stocks = new HashMap<Stock,Double>();
                @Override
                protected void setup(Context context) throws IOException, InterruptedException {
                        stockSymbol = context.getConfiguration().get("stockSymbol");
                        Configuration conf = context.getConfiguration();
                        //Path[] files = context.getLocalCacheFiles();
                        Path[] files = DistributedCache.getLocalCacheFiles(conf);
                        FileSystem fs = FileSystem.getLocal(conf);
                        BufferedReader br = new BufferedReader(new InputStreamReader(fs.open(files[0])));
                        String[] splits = br.readLine().split(",");
                        if(splits[1].equals(stockSymbol)){
                                stocks.put(new Stock(splits[1],splits[2]), Double.valueOf(splits[3]));
                        }
                }

                @Override
                protected void map(LongWritable key, Text value, Context context)
                                throws IOException, InterruptedException {
                        Stock tmp = null;
                        String currentLine = value.toString();
                       
                        String[] splits = currentLine.split(",");
                        if(splits[1].equals(stockSymbol)){
                                context.getCounter(TESTW.HELLO).increment(1);
                                context.getCounter(TESTW.MISSING).increment(stocks.size());
                                tmp = new Stock(splits[1],splits[2]);
                        }
                       
                        if(stocks.containsKey(tmp)){
                                context.write(tmp, new StockPrices(stocks.get(tmp), Double.valueOf(splits[6])));
                        }
                }
        }

        public int run(String[] args) throws Exception {
                Job job = Job.getInstance(getConf(), "MapSideJoinJob");
                job.setJarByClass(getClass());
                Configuration conf = job.getConfiguration();
                conf.set("stockSymbol", args[0]);
                conf.set(TextOutputFormat.SEPERATOR, ",");
               
                Path out = new Path("joinoutput");
                out.getFileSystem(conf).delete(out,true);
                FileInputFormat.setInputPaths(job, new Path("stocks"));
                FileOutputFormat.setOutputPath(job, out);
                DistributedCache.addCacheFile(new Path("dividends/NYSE_dividends_A.csv").toUri() , conf);
                //job.addCacheFile(new Path("dividends/NYSE_dividends_A.csv").toUri());
                job.setMapperClass(MapSideJoinMapper.class);
                job.setInputFormatClass(TextInputFormat.class);
               
                job.setOutputFormatClass(TextOutputFormat.class);
                job.setOutputKeyClass(Text.class);
                job.setOutputValueClass(Text.class);
                job.setMapOutputKeyClass(Stock.class);
                job.setMapOutputValueClass(StockPrices.class);
               
                job.setNumReduceTasks(0);
               
                return job.waitForCompletion(true)?0:1;
        }

        public static void main(String[] args) {
                int result = 0;
                try {
                        result = ToolRunner.run(new Configuration(),  new MapSideJoin(), args);
                } catch (Exception e) {
                        e.printStackTrace();
                }
                System.exit(result);
        }
}

nickpro 发表于 2014-9-27 08:04
package hadoop.mapjoin;

import hadoop.other.Stock;

DistributedCache.addCacheFile(new Path("/dividends/NYSE_dividends_A.csv").toUri() , conf);
前面加上斜杠/

howtodown 发表于 2014-9-27 11:20
DistributedCache.addCacheFile(new Path("/dividends/NYSE_dividends_A.csv").toUri() , conf);
前面加 ...

加上“/”肯定出错啊，我这个文件又不是放在根目录里面

虽然没有做过，但是感觉你的conf配置有问题，你可以先写一个简单的例子，然后在做这个复杂的例子。

sstutu 发表于 2014-9-27 11:49
虽然没有做过，但是感觉你的conf配置有问题，你可以先写一个简单的例子，然后在做这个复杂的例子。

已经写过好几个例子了，没有出错，就是做到这个例子的时候出错了

nickpro 发表于 2014-9-27 11:51
已经写过好几个例子了，没有出错，就是做到这个例子的时候出错了

你可以对比这个例子，跟其他的例子的区别在什么地方，如果找不出来，你可以在原先的例子上，不断修改，逐渐改成你的这个例子。就能找到问题所在了。

sstutu 发表于 2014-9-27 11:53
你可以对比这个例子，跟其他的例子的区别在什么地方，如果找不出来，你可以在原先的例子上，不断修改，逐 ...

http://www.codelast.com/?p=8131   这是我在网上找的，你看一下，该怎么解决？

sstutu 发表于 2014-9-27 11:53
你可以对比这个例子，跟其他的例子的区别在什么地方，如果找不出来，你可以在原先的例子上，不断修改，逐 ...

http://www.codelast.com/?p=8131  这是我在网上找到的，他说使用旧api可以成功，不过我试了一下，也没法成功啊