int stage1NumTasks = genSplitsInAM ? -1 : inputSplitInfo.getNumTasks();
Class<? extends TezRootInputInitializer> inputInitializerClazz =
genSplitsInAM ? (initializerClass == null ? MRInputAMSplitGenerator.class : initializerClass)
: null;
LOG.info("Using initializer class: " + initializerClass);
Vertex stage1Vertex = new Vertex("map", new ProcessorDescriptor(
MapProcessor.class.getName()).setUserPayload(stage1Payload),
stage1NumTasks, Resource.newInstance(256, 1));
MRHelpers.addMRInput(stage1Vertex, stage1InputPayload, inputInitializerClazz);
Vertex stage2Vertex = new Vertex("ireduce", new ProcessorDescriptor(
ReduceProcessor.class.getName()).setUserPayload(
MRHelpers.createUserPayloadFromConf(stage2Conf)),
1, Resource.newInstance(256, 1));
Vertex stage3Vertex = new Vertex("reduce", new ProcessorDescriptor(
ReduceProcessor.class.getName()).setUserPayload(stage3Payload),
1, Resource.newInstance(256, 1));
MRHelpers.addMROutputLegacy(stage3Vertex, stage3Payload);
Map<String, LocalResource> commonLocalResources = new HashMap<String, LocalResource>();