/**
 * Add the JAR at the given HDFS path to the Classpath
 *
 * @param conf
 * @param path
 */
public static void addJarToJobClasspath(JobConf conf, String path) {
    addJarsToJobClasspath(conf, new String[] { path });
}

/**
 * Add the JARs from the given HDFS paths to the Classpath
 *
 * @param conf
 * @param paths
 */
public static void addJarsToJobClasspath(JobConf conf, String[] paths) {

    URL[] libjars = new URL[paths.length];

    // Grab JARs from HDFS and cache locally
    for (int i = 0; i < paths.length; i++) {
        try {
            String hdfsPath = paths[i];
            Path srcPath = new Path(hdfsPath);

            FileSystem fs = FileSystem.get(conf);
            File dst = File.createTempFile(srcPath.getName() + "-", ".jar");
            dst.deleteOnExit();
            Path dstPath = new Path(dst.getAbsolutePath());

            fs.copyToLocalFile(srcPath, dstPath);
            libjars[i] = new URL("file:" + dst.getAbsolutePath());

        } catch (IOException ex) {
            throw new RuntimeException("Error setting up classpath", ex);
        }
    }

    // Set classloader in current conf/thread
    conf.setClassLoader(
            new URLClassLoader(libjars, conf.getClassLoader()));

    Thread.currentThread().setContextClassLoader(
            new URLClassLoader(libjars, Thread.currentThread().getContextClassLoader()));

    // Append to tmpjars variable
    String jars = conf.get("tmpjars");
    if (jars == null) {
        jars = "";
    }
    for (int i = 0; i < libjars.length; i++) {
        URL url = libjars[i];
        if (jars.length() > 0) {
            jars += ",";
        }
        jars += url.toString();
    }
    conf.set("tmpjars", jars);

}