Hi
My code creates a new job named "job 1" which writes something to distributed
cache (say a text file) and the job gets completed.
Just to manage expectations, you add files to the distributed cache_in the job
driver_, and the framework makes them available to maps and reducers.
Adding a file to the distributed cache in a mapper and expecting to have it
available in other mappers/reducers or later jobs is not a supported use case.
If you want a mapper/reducer to send info back to the job driver, then use
counters or write regular files (in the job output dir).
If you want to add a file to the distributed cache and have multiple jobs use
it, then:
Note that DistributedCache.add*(Path, Conf) append the file name to internal
conf properties. (then when you submit the job, the framework will read those
properties to find out what files need to be distributed, added to the
classpath, etc.).
So DistributedCache.addCacheFile(*f*, *conf1*) will write something into
*conf1*. When you launch *job1* using *conf1*, *f* gets distributed. When you
launch *job2* using *conf2*, then *job2* may not distribute file *f*, depending
on how you build *conf2* (independent of *conf1*, or a copy/clone of *conf1*).
hth
Gabriel Balan
P.S. Btw, you don't have to copy the local file to hdfs using "IOUtils.copyBytes(in,
out, 4096, true);"
Try FileSystem.copyFromLocalFile
<https://hadoop.apache.org/docs/r2.6.1/api/src-html/org/apache/hadoop/fs/FileSystem.html#line.1885>
or "hadoop -jar foo.jar MyClass -file /home/siddharth/Desktop/data/bloom_filter"
On 6/13/2016 5:14 AM, Siddharth Dawar wrote:
Hi Jeff,
Thanks for your prompt reply. Actually my problem is as follows:
My code creates a new job named "job 1" which writes something to distributed
cache (say a text file) and the job gets completed.
Now, I want to create some n number of jobs in while loop below, which reads the text file written
by "job 1" from the distributed cache. So my question is, "*How to share content
among multiple jobs using distributed cache*" ?
*Another part of the problem *is that I don't know how to get instance of
running job from
JobClient.runJob(conf2);
*so that I can use job.addcachefiles(..) command/*
**
while (true)
{
JobConf conf2 = new JobConf(getConf(),graphMining.class);
conf2.setJobName("sid");
conf2.setMapperClass(mapperMiner.class);
conf2.setReducerClass(reducerMiner.class);
conf2.setInputFormat(SequenceFileInputFormat.class);
conf2.setOutputFormat(SequenceFileOutputFormat.class);
conf2.setOutputValueClass(BytesWritable.class);
conf2.setMapOutputKeyClass(Text.class);
conf2.setMapOutputValueClass(MapWritable.class);
conf2.setOutputKeyClass(Text.class);
conf2.setNumMapTasks(Integer.parseInt(args[3]));
conf2.setNumReduceTasks(Integer.parseInt(args[4]));
FileInputFormat.addInputPath(conf2, new Path(input));
FileOutputFormat.setOutputPath(conf2, new Path(output)); }
RunningJob job = JobClient.runJob(conf2); }
On Wed, Jun 8, 2016 at 3:50 AM, Guttadauro, Jeff <jeff.guttada...@here.com
<mailto:jeff.guttada...@here.com>> wrote:
Hi, Siddharth.
I was also a bit frustrated at what I found to be scant documentation on
how to use the distributed cache in Hadoop 2. The DistributedCache class
itself was deprecated in Hadoop 2, but there don’t appear to be very clear
instructions on the alternative. I think it’s actually much simpler to work
with files on the distributed cache in Hadoop 2. The new way is to add files
to the cache (or cacheArchive) via the Job object:
job.addCacheFile(/uriForYourFile/)
job.addCacheArchive(/uriForYourArchive/);
The cool part is that, if you set up your URI so that it has a
“#/yourFileReference/” at the end, then Hadoop will set up a symbolic link
named “/yourFileReference/” in your job’s working directory, which you can use
to get at the file or archive. So, it’s as if the file or archive is in the
working directory. That obviates the need to even work with the
DistributedCache class in your Mapper or Reducer, since you can just work with
the file (or path using nio) directly.
Hope that helps.
-Jeff
*From:*Siddharth Dawar [mailto:siddharthdawa...@gmail.com
<mailto:siddharthdawa...@gmail.com>]
*Sent:* Tuesday, June 07, 2016 4:06 AM
*To:* user@hadoop.apache.org <mailto:user@hadoop.apache.org>
*Subject:* Accessing files in Hadoop 2.7.2 Distributed Cache
Hi,
I want to use the distributed cache to allow my mappers to access data in
Hadoop 2.7.2. In main, I'm using the command
String hdfs_path="hdfs://localhost:9000/bloomfilter";
InputStream in = new BufferedInputStream(new
FileInputStream("/home/siddharth/Desktop/data/bloom_filter"));
Configuration conf = new Configuration();
fs = FileSystem.get(java.net.URI.create(hdfs_path), conf);
OutputStream out = fs.create(new Path(hdfs_path));
//Copy file from local to HDFS
IOUtils.copyBytes(in, out, 4096, true);
System.out.println(hdfs_path + " copied to
HDFS");DistributedCache.addCacheFile(new Path(hdfs_path).toUri(), conf2);
DistributedCache.addCacheFile(new Path(hdfs_path).toUri(), conf2);
The above code adds a file present on my local file system to HDFS and adds
it to the distributed cache.
However, in my mapper code, when I try to access the file stored in
distributed cache, the Path[] P variable gets null value. d
public void configure(JobConf conf)
{
this.conf = conf;
try {
Path []
p=DistributedCache.getLocalCacheFiles(conf);
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
Even when I tried to access distributed cache from the following code
in my mapper, the code returns the error that bloomfilter file doesn't exist
strm = new DataInputStream(new FileInputStream("bloomfilter"));
// Read into our Bloom filter.
filter.readFields(strm);
strm.close();
However, I read somewhere that if we add a file to distributed cache, we
can access it
directly from its name.
Can you please help me out ?
--
The statements and opinions expressed here are my own and do not necessarily
represent those of Oracle Corporation.