6
\$\begingroup\$

Just got done implementing a shortest path algo of a unit cost DAG BFS in Hadoop and wanting to get anyone's opinion on how I could have done it better. The main issue I see is that I lose the parallel nature of hadoop by forcing the mappers to wait for the output of the previous mapper before beginning.

The thing works and finds the shortest path given two input files: nodes.txt and edges.txt.

nodes.txt represents all the nodes in the DAG and is of the form

1 "A"
2 "B"
3 "C"
4 "D"
5 "E"
6 "F"
7 "G"
8 "Z"
9 "X"

edges.txt represents all the edges of the DAG and is of the form

1 2
1 3
1 4
2 5
2 6
3 5
3 6
4 7
5 8
6 8
7 8
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class ShortestPathDAGBFS
{

    public static class Map extends Mapper<LongWritable, Text, Text, Text>
    {
        HashMap<String, String> nodes = new HashMap<>();
        HashMap<String, LinkedList<String>> edges = new HashMap<>();
        String sourceNodeId;
        String destNodeId;
        Pattern pattern = Pattern.compile("[^\\S\\n]{1,4}");
        boolean found = false;

        @Override
        public void setup(Context context) throws IOException
        {
            nodes = GlobalFunctions.parseNodesFile(context.getConfiguration().get("Nodes"));
            edges = GlobalFunctions.parseEdgesFile(context.getConfiguration().get("Edges"));
            sourceNodeId = GlobalFunctions.findNodeId(context.getConfiguration().get("Source"), nodes).trim();
            destNodeId = GlobalFunctions.findNodeId(context.getConfiguration().get("Dest"), nodes).trim();
        }

        @Override
        public void map(LongWritable offset, Text line, Context context) throws IOException, InterruptedException
        {
            String[] edge = pattern.split(line.toString());
            edge[0] = edge[0].trim();
            edge[1] = edge[1].trim();
            LinkedList<String> outNodes = edges.get(edge[1]);
            String newPath = edge[0] + "->" + nodes.get(edge[1]);
            if (outNodes != null && !found)
            {
                for (String nextNodeId : outNodes)
                {
                    context.write(new Text(newPath.toString()), new Text(nextNodeId.toString()));
                    if (nextNodeId.toString().equalsIgnoreCase(destNodeId))
                    {
                        context.getCounter("FoundDest", "FoundDest").increment(1);
                        found= true;
                        context.write(new Text(edge[0].toString()), new Text(edge[1].toString()));
                        return;
                    }
                }
            }
        }
    }

    public static class Reduce extends Reducer<Text, Text, Text, Text>
    {
        String destNodeId;
        HashMap<String, String> nodes = new HashMap<>();
        boolean written = false;

        @Override
        public void setup(Context context) throws IOException
        {
            nodes = GlobalFunctions.parseNodesFile(context.getConfiguration().get("Nodes"));
            destNodeId = GlobalFunctions.findNodeId(context.getConfiguration().get("Dest"), nodes);
        }

        @Override
        public void reduce(Text prevPath, Iterable<Text> values, Context context) throws IOException, InterruptedException
        {
            for (Text newestNode : values)
            {
                if (newestNode.toString().trim().equals(destNodeId) && !written)
                {
                    String destNode = context.getConfiguration().get("Dest");
                    String path = prevPath.toString().replace("->", " ");
                    path += " " + destNode;
                    context.write(new Text(path), null);
                    written = true;
                }
                else if (!written)
                {
                    context.write(new Text(prevPath.toString()), new Text(newestNode.toString()));
                    System.out.println("Reducer output: " + prevPath.toString() + " " + newestNode.toString());
                }
            }

        }
    }

    public static class GlobalFunctions
    {

        public static HashMap<String, String> parseNodesFile(String nodesPath) throws IOException
        {
            HashMap<String, String> nodes = new HashMap<>();
            Path path = new Path(nodesPath);
            FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
            InputStream is = (fs.open(new Path(path.toUri().toString())));
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, Charset.defaultCharset())))
            {
                String line;
                while ((line = reader.readLine()) != null)
                {
                    String[] node = line.replace(' ', '\0').split("\"");
                    nodes.put(node[0].trim(), node[1].trim());
                }
            }
            return nodes;
        }

        public static HashMap<String, LinkedList<String>> parseEdgesFile(String edgesPath) throws IOException
        {
            HashMap<String, LinkedList<String>> edges = new HashMap<>();
            Path path = new Path(edgesPath);
            FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
            InputStream is = (fs.open(new Path(path.toUri().toString())));
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, Charset.defaultCharset())))
            {
                String line;
                Pattern pattern = Pattern.compile("[^\\S\\n]{1,4}");
                while ((line = reader.readLine()) != null)
                {
                    String[] edge = pattern.split(line);
                    LinkedList<String> adjacencyList = edges.get(edge[0]);
                    if (adjacencyList == null && !edge[0].equals(edge[1]))
                    {
                        LinkedList<String> list = new LinkedList<>();
                        list.add(edge[1]);
                        edges.put(edge[0], list);
                    }
                    else if (!edge[0].equals(edge[1]))
                    {
                        adjacencyList.add(edge[1]);
                    }
                }
            }
            return edges;
        }

        public static String findNodeId(String source, HashMap<String, String> nodes)
        {
            for (Entry<String, String> e : nodes.entrySet())
            {
                if (e.getValue().equals(source))
                {
                    return e.getKey();
                }
            }
            return null;
        }
    }

    public static class FirstMap extends Mapper<LongWritable, Text, Text, Text>
    {

        HashMap<String, String> nodes = new HashMap<>();
        Pattern pattern = Pattern.compile("[^\\S\\n]{1,4}");
        String sourceNodeId;

        @Override
        public void setup(Context context) throws IOException
        {
            nodes = GlobalFunctions.parseNodesFile(context.getConfiguration().get("Nodes"));
            sourceNodeId = GlobalFunctions.findNodeId(context.getConfiguration().get("Source"), nodes).trim();
        }

        @Override
        public void map(LongWritable offset, Text line, Context context) throws IOException, InterruptedException
        {
            String[] edge = pattern.split(line.toString());
            edge[0] = edge[0].trim();
            if (sourceNodeId.equals(edge[0]))
            {
                context.write(new Text(nodes.get(edge[0])), new Text(edge[1]));
                System.out.println("Writing for first iteration" + edge[0] + " " + edge[1]);
            }
        }
    }

    public static void main(String[] args) throws Exception
    {
        GenericOptionsParser parser = new GenericOptionsParser(args);
        String[] arguments = parser.getRemainingArgs();
        if (arguments.length != 5)
        {
            usage(System.out);
            System.exit(-1);
        }
        Configuration conf = parser.getConfiguration();
        conf.set("Source", arguments[3]);
        conf.set("Dest", arguments[4]);
        conf.set("Nodes", arguments[0]);
        conf.set("Edges", arguments[1]);

        Job job = new Job(conf);
        job.setJarByClass(ProblemIII.class);
        job.setMapperClass(FirstMap.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        FileInputFormat.addInputPaths(job, arguments[1]);
        FileOutputFormat.setOutputPath(job, new Path("tmpout-part0"));
        job.waitForCompletion(true);

        boolean found;
        long outputBytes;
        int iteration = 0;
        do
        {
            job = initJob(conf);
            FileInputFormat.addInputPath(job, new Path("tmpout-part" + iteration));
            FileOutputFormat.setOutputPath(job, new Path("tmpout-part" + (iteration + 1)));
            job.waitForCompletion(true);
            found = job.getCounters().findCounter("FoundDest", "FoundDest").getValue() == 1 ? true : false;
            outputBytes = job.getCounters().findCounter("org.apache.hadoop.mapred.Task$Counter", "MAP_OUTPUT_BYTES").getValue();
            ++iteration;
        } while (!found && outputBytes != 0);
        if (outputBytes == 0)
        {
            FileSystem fs = FileSystem.get(new Configuration());
            FSDataOutputStream file = FileSystem.get(new Configuration()).create(new Path(arguments[2]+"/output.txt"));
            file.writeChars(arguments[3] + " null " + arguments[4]);
            file.close();
            System.exit(0);
        } 
        else
        {
            Job finalJob = initJob(conf);
            FileInputFormat.addInputPaths(finalJob, "tmpout-part" + (iteration));
            FileOutputFormat.setOutputPath(finalJob, new Path(arguments[2]));
            System.exit(finalJob.waitForCompletion(true) ? 0 : 1);
        }
    }

    private static void usage(PrintStream out)
    {
        out.println("usage: ProblemIII <nodes path> <edges path> <output path> <source protein> <destination protein>");
    }

    private static Job initJob(Configuration conf) throws IOException
    {
        Job job = new Job(conf);
        job.setJarByClass(ProblemIII.class);
        job.setMapperClass(Map.class);
        job.setReducerClass(Reduce.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        return job;
    }
}
\$\endgroup\$
2
\$\begingroup\$

I'm not familiar with Hadoop, take my advice with care.

  1. Are you sure that you want to use the default charset?

    try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, Charset.defaultCharset()))) {
    

    It could vary from system to system. Usually it's a good idea using UTF-8 explicitly.

  2. Since both newPath and nextNodeId are Strings,

    context.write(new Text(newPath.toString()), new Text(nextNodeId.toString()));
    

    could be

    context.write(new Text(newPath), new Text(nextNodeId));
    

    It's the same, since String.toString has the following implementation:

    public String toString() {
        return this;
    }
    

    (There are other similar occurrences in the code, find them.)

  3. As far as I see Text has a constructor which accepts Text parameters. You might want to use that instead of converting it back to String:

    context.write(new Text(prevPath.toString()), new Text(newestNode.toString()));
    
  4. The setup method also queries Dest. You might want to store that in a field or move it outside the loop:

    for (Text newestNode: values) {
        if (newestNode.toString().trim().equals(destNodeId) && !written) {
            String destNode = context.getConfiguration().get("Dest");
    
  5. You could invert the conditon in the map method to make the code flatten:

    @Override
    public void map(LongWritable offset, Text line, Context context) 
            throws IOException, InterruptedException {
        String[] edge = pattern.split(line.toString());
        edge[0] = edge[0].trim();
        edge[1] = edge[1].trim();
        List<String> outNodes = edges.get(edge[1]);
        String newPath = edge[0] + "->" + nodes.get(edge[1]);
        if (outNodes == null || found) {
            return;
        }
        for (String nextNodeId: outNodes) {
            context.write(new Text(newPath), new Text(nextNodeId));
            if (nextNodeId.toString().equalsIgnoreCase(destNodeId)) {
                context.getCounter("FoundDest", 
                    "FoundDest").increment(1);
                found = true;
                context.write(new Text(edge[0].toString()), 
                    new Text(edge[1].toString()));
                return;
            }
        }
    }
    
  6. Actually, I guess the found check could be right at the beginning:

    @Override
    public void map(LongWritable offset, Text line, Context context) 
            throws IOException, InterruptedException {
        if (found) {
            return;
        }
        ...
    }
    
  7. The reduce method also could use a guard clause. If I'm right it could even break the loop since if written is true both nested conditions will be false.

    @Override
    public void reduce(Text prevPath, 
            Iterable<Text> values, Context context) 
            throws IOException, InterruptedException {
        for (Text newestNode: values) {
            if (written) {
                break;
            }
            ...
        }
    }
    
  8. sourceNodeId = GlobalFunctions.findNodeId(context.getConfiguration().get("Source"), nodes).trim();
    

    findNodeId could be a little bit smarter. All of the four uses of it trim its output and pass an object which is get from context.getConfiguration().get("somekey"). A wrapper like this could eliminate some duplication:

    public static String findTrimmedNodeId(
            Configuration configuration, 
            String key, Map<String, String> nodes) {
        return findNodeId(configuration.get(key), nodes).trim();
    }
    
  9. The fs variable is unused here:

    FileSystem fs = FileSystem.get(new Configuration());
    

    Are you sure that these statements are required?

  10. I'd change ShortestPathDAGBFS to ShortestPathDagBfs

    From Effective Java, 2nd edition, Item 56: Adhere to generally accepted naming conventions:

    While uppercase may be more common, a strong argument can made in favor of capitalizing only the first letter: even if multiple acronyms occur back-to-back, you can still tell where one word starts and the next word ends. Which class name would you rather see, HTTPURL or HttpUrl?

  11. Calling the mapper Map is confusing. I'd choose DagBfsMapper or something descriptive. The same is true for Reduce.

  12. HashMap<...> reference types should be simply Map<...>, as well as LinkedList<...> could be List<...> (Effective Java, 2nd edition, Item 52: Refer to objects by their interfaces)

  13. Consider using LineIterator, it would simplify the code a little bit.

  14. Try to use more descriptive variable names:

    Pattern pattern = Pattern.compile("[^\\S\\n]{1,4}");
    

    What is this pattern supposed to match?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.