throws NotFoundException, InstantiationException, IllegalAccessException,
CannotCompileException {
GraphParser graphparser = (GraphParser) graphParserClass().newInstance();
EdgeTransformMR mr = new EdgeTransformMR(reduceEndPoint, graphparser,
ParserFactory.getParserByClassName(vidClass().getName()),
ParserFactory.getParserByClassName(edataClass().getName()));
Class valueClass = EdgeTransformJobValueFactory
.getValueClassByClassName(vidClass().getName(), edataClass().getName());
mr.setKeyValueClass(vidClass(), valueClass);
// Distribute new class file to cluster.
FsUtil.distributedTempClassToClassPath(mr.getConf());
mr.setFunctionClass(reduceFunction().getClass(), applyFunction().getClass());
if (userOpts != null) {
mr.setUserOptions(userOpts);
}
try {
mr.run(input, output);
} catch (Exception e) {
e.printStackTrace();
return false;
}
return true;