hadoop-mapreduce-dev mailing list archives

Philippe created MAPREDUCE-4688:
-----------------------------------

             Summary: setJarByClass does not work under JBoss AS 7
                 Key: MAPREDUCE-4688
                 URL: https://issues.apache.org/jira/browse/MAPREDUCE-4688
             Project: Hadoop Map/Reduce
          Issue Type: Bug
    Affects Versions: 1.0.3, 0.20.2
         Environment: Hadoop Cloudera CDH3 Cluster w/Hadoop 0.20.2 on CentOS 5/6
Client on JBoss AS 7.1.1.Final CentOS/Windows
            Reporter: Philippe
            Priority: Minor


Hello,

I’m using Hadoop as a client from a J2EE web application. One of the lib within my EAR is
a jar containing several Map/Reduce jobs. Using JBoss AS 4 in the past, I had no problem running
the jobs with the following code:
{code}
try {
    final Configuration conf = HBaseConfiguration.create();
    // Load all Hadoop configuration
    conf.addResource("core-site.xml");
    conf.addResource("hdfs-site.xml");
    conf.addResource("mapred-site.xml");
    conf.addResource("hbase-site.xml");

    final Job job = new Job(conf, "My Job");
    job.setJarByClass(MyJobClass.getClass());
			
    			
    TableMapReduceUtil.initTableMapperJob(...);
			
    TableMapReduceUtil.initTableReducerJob(...);
			
    final boolean status = job.waitForCompletion(true);
} ...
{code}

Since, we have moved to JBoss AS 7 and the method setJarByClass is no longer working. Indeed,
in *org.apache.hadoop.mapred.JobConf.findContainingJar(Class)* the retrieved URL does not
have a *jar* protocol, but a JBoss *vfs* protocol. So, it always return null and the jar is
not sent to the Map/Reduce cluster.
With the VFS protocol the resource name may be or may not be the actual system file name of
the resource. I mean, the class file is within the jar file which may be within an ear file
in case of a non-exploded deployment, so there are not system File corresponding to the resource.
Though, I guess similar issues may happen with jar: protocol.

In order to make the job working with JBoss AS-7, I did the following implementation of the
Job class. This override the setJarByClass mechanism, by creating a temporary jar file from
the actual jar file read from vfs.

{code}
/**
 * Patch of Map/Red Job to handle VFS jar file
 */
public class VFSJob extends Job
{
	
    /** Logger */
    private static final transient Logger logger = LoggerFactory.getLogger(VFSJob.class);

    public VFSJob() throws IOException {
        super();
    }

    public VFSJob(Configuration conf) throws IOException {
        super(conf);
    }

    public VFSJob(Configuration conf, String jobName) throws IOException {
        super(conf, jobName);
    }
	
    private File temporaryJarFile;

    /**
     * Patch of setJarByClass to handle VFS
     */
    @Override
    public void setJarByClass(Class<?> cls) {
       final ClassLoader loader = cls.getClassLoader();
        final String classFile = cls.getName().replaceAll("\\.", "/") + ".class";
        JarInputStream is = null;
        JarOutputStream os = null;
        try {
            final Enumeration<URL> itr = loader.getResources(classFile);
            while (itr.hasMoreElements()) {
                final URL classUrl = itr.nextElement();
                // This is the trick
                if (!"vfs".equals(classUrl.getProtocol())) {
                    continue;
                }
			   
                final String jarFile = classUrl.getFile().substring(0, classUrl.getFile().length()
- (classFile.length() + 1) ); //+1 because of '/'
                final URL jarUrl = new URL(classUrl.getProtocol(), classUrl.getHost(), classUrl.getPort(),
jarFile, new org.jboss.vfs.protocol.VirtualFileURLStreamHandler());
			  
                temporaryJarFile = File.createTempFile("mapred", ".jar");

                is = (JarInputStream) jarUrl.openStream();
                os = new JarOutputStream(new FileOutputStream(temporaryJarFile));

                final byte[] buffer = new byte[2048];

                for (JarEntry entry = is.getNextJarEntry(); entry !=null; entry = is.getNextJarEntry())
{
                    os.putNextEntry(entry);
                    int bytesRead;
                    while ((bytesRead = is.read(buffer)) != -1) {
                        os.write(buffer, 0, bytesRead);
                    }
                }

                this.conf.setJar(temporaryJarFile.getPath());
                return;
            }
        } catch (IOException e) {
            throw new RuntimeException(e);
        } finally {
            if (is != null) {
                try {
                    is.close();
                } catch (IOException e) {
                    logger.error("Error closing input stream", e);
                }
            }
            if (os != null) {
                try {
                    os.close();
                } catch (IOException e) {
                    logger.error("Error closing output stream", e);
                }
            }
        }
		
        return;
    }
	
    /**
     * Clean the temporary jar file created
     */
    public void clean() {
        if (temporaryJarFile != null && temporaryJarFile.exists()) {
            final boolean isDeleted = temporaryJarFile.delete();
            if (!isDeleted) {
                logger.error("Error while deleting temporary jar " + temporaryJarFile);
            }
        }
    }


}
{code}

So, after the call, I must not forget deleting the temporary jar file.
{code}
VFSJob job = null;
try {
    final Configuration conf = HBaseConfiguration.create();
    // Load all Hadoop configuration
    conf.addResource("core-site.xml");
    conf.addResource("hdfs-site.xml");
    conf.addResource("mapred-site.xml");
    conf.addResource("hbase-site.xml");

    job = new VFSJob(conf, "My Job");
    job.setJarByClass(MyJobClass.getClass());
			
			
    TableMapReduceUtil.initTableMapperJob(...);
			
    TableMapReduceUtil.initTableReducerJob(...);
			
    final boolean status = job.waitForCompletion(true);
} ...
} finally {
    job.clean();
}
{code}








--
This message is automatically generated by JIRA.
If you think it was sent incorrectly, please contact your JIRA administrators
For more information on JIRA, see: http://www.atlassian.com/software/jira