Hi Nick,
I am not able to start the following job. I have the file that has to be
passed to distributedcache in the local filesystem of the task tracker.
Can you tell me if I am missing something?
import org.apache.hadoop.fs.*;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.mapred.lib.*;
import org.apache.hadoop.util.*;
import org.apache.hadoop.filecache.*;
import java.io.*;
import java.util.*;
import java.text.SimpleDateFormat;
import java.net.*;
public class Test extends Configured
{
public static class MapClass extends MapReduceBase implements
Mapper<LongWritable,Text,Text,Text>
{
private FileSystem fs;
private Path[] localFiles;
private String str;
public void configure(JobConf job)
{
try{
fs = FileSystem.getLocal(new Configuration());
localFiles = DistributedCache.getLocalCacheFiles(job);
}
catch(IOException e){
System.out.println("Exception while getting cached files");
}//catch(IOException e){
}//public void configure(JobConf job)
public void map(LongWritable Key,Text value,OutputCollector<Text,Text>
output,Reporter reporter) throws IOException
{
BufferedReader readBuffer = new BufferedReader(new
FileReader(localFiles[0].toString()));
str = readBuffer.readLine();
output.collect(new Text(str),new Text(str));
}//public void map(LongWritable Key,Text
value,OutputCollector<Text,Text> output,Reporter reporter) throws
IOException
public void close() throws IOException
{
fs.close();
}//public void close() throws IOException
}//public static class MapClass extends MapReduceBase implements Mapper<>
public static class ReduceClass extends MapReduceBase implements
Reducer<Text,Text,Text,Text>
{
public void reduce(Text key,Iterator<Text>
values,OutputCollector<Text,Text> output,Reporter reporter) throws
IOException
{
}//public void reduce(Text key,Iterator<Text>
values,OutputCollector<Text,Text> output,Reporter reporter) throws
IOException
}//public static class ReduceClass extends MapReduceBase implements
Reducer<Text,Text,Text,Text>
public static void main(String[] args)
{
JobConf conf = new JobConf(Test.class);
JobClient client = new JobClient();
conf.setMapperClass(Test.MapClass.class);
conf.setReducerClass(IdentityReducer.class);
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(Text.class);
conf.setInputPath(new Path("input"));
conf.setOutputPath(new Path("output"));
try{
DistributedCache.addCacheFile(new
URI("/home/udaya/hadoop-0.18.3/file_to_distribute"), conf);
}
catch(URISyntaxException e)
{}
try{
JobClient.runJob(conf);
}
catch(Exception e)
{
System.out.println("Error starting the job");
}
}//public static void main(String[] args)
}//public class Test extends Configured implements Tools
On Wed, Feb 3, 2010 at 7:27 PM, Nick Jones <[email protected]> wrote:
> Hi Udaya,
> The following code uses already existing cache files as part of the map to
> process incoming data. I apologize on the naming conventions, but the code
> had to be stripped. I also removed several variable assignments, etc..
>
> public class MySpecialJob {
> public static class MyMapper extends MapReduceBase implements
> Mapper<LongWritable, MyMapInputValueClass, MyMapOutputKeyClass,
> BigIntegerWritable> {
>
> private Path[] dcfiles;
> ...
>
> public void configure(JobConf job) {
> // Load cached files
> dcfiles = new Path[0];
> try {
> dcfiles = DistributedCache.getLocalCacheFiles(job);
> } catch (IOException ioe) {
> System.err.println("Caught exception while getting cached files: " +
> StringUtils.stringifyException(ioe));
> }
> }
>
> public void map(LongWritable key, MyMapInputValueClass value,
> OutputCollector<MyMapOutputKeyClass, BigIntegerWritable> output,
> Reporter reporter) throws IOException {
> ...
> for (Path dcfile : dcfiles) {
> if(dcfile.getName().equalsIgnoreCase(file_match)) {
> readbuffer = new BufferedReader(
> new FileReader(dcfile.toString()));
> ...
> while((raw_line = readbuffer.readLine()) != null) {
> ...
>
> public static void main(String[] args) throws Exception {
> JobConf conf = new JobConf(MySpecialJob.class);
> ...
> DistributedCache.addCacheFile(new URI("/path/to/file1.txt"), conf);
> DistributedCache.addCacheFile(new URI("/path/to/file2.txt"), conf);
> DistributedCache.addCacheFile(new URI("/path/to/file3.txt"), conf);
> ...
> }
> }
>
> Nick Jones
>
>
>
> Udaya Lakshmi wrote:
>
>> Hi,
>> As a newbie to hadoop, I am not able to figure out how to use
>> DistributedCache class. Can someone give me a small code which
>> distributes file to the cluster and the show how to open and use the
>> file in the map or reduce task.
>> Thanks,
>> Udaya
>>
>>
>
