@Test public void testMapProcessor() throws Exception { String dagName = "mrdag0"; String vertexName = MultiStageMRConfigUtil.getInitialMapVertexName(); JobConf jobConf = new JobConf(defaultConf); setUpJobConf(jobConf); MRHelpers.translateVertexConfToTez(jobConf); jobConf.setInt(MRJobConfig.APPLICATION_ATTEMPT_ID, 0); jobConf.setBoolean(MRJobConfig.MR_TEZ_SPLITS_VIA_EVENTS, false); jobConf.set(MRFrameworkConfigs.TASK_LOCAL_RESOURCE_DIR, new Path(workDir, "localized-resources").toUri().toString()); Path mapInput = new Path(workDir, "map0"); MapUtils.generateInputSplit(localFs, workDir, jobConf, mapInput); InputSpec mapInputSpec = new InputSpec("NullSrcVertex", new InputDescriptor(MRInputLegacy.class.getName()) .setUserPayload(MRHelpers.createMRInputPayload(jobConf, null)), 1); OutputSpec mapOutputSpec = new OutputSpec("NullDestVertex", new OutputDescriptor(LocalOnFileSorterOutput.class.getName()) .setUserPayload(TezUtils.createUserPayloadFromConf(jobConf)), 1); LogicalIOProcessorRuntimeTask task = MapUtils.createLogicalTask(localFs, workDir, jobConf, 0, new Path(workDir, "map0"), new TestUmbilical(), dagName, vertexName, Collections.singletonList(mapInputSpec), Collections.singletonList(mapOutputSpec)); task.initialize(); task.run(); task.close(); TezInputContext inputContext = task.getInputContexts().iterator().next(); TezTaskOutput mapOutputs = new TezLocalTaskOutputFiles(jobConf, inputContext.getUniqueIdentifier()); Path mapOutputFile = mapOutputs.getInputFile(new InputAttemptIdentifier(0, 0)); LOG.info("mapOutputFile = " + mapOutputFile); IFile.Reader reader = new IFile.Reader(localFs, mapOutputFile, null, null, null, false, 0, -1); LongWritable key = new LongWritable(); Text value = new Text(); DataInputBuffer keyBuf = new DataInputBuffer(); DataInputBuffer valueBuf = new DataInputBuffer(); long prev = Long.MIN_VALUE; while (reader.nextRawKey(keyBuf)) { reader.nextRawValue(valueBuf); key.readFields(keyBuf); value.readFields(valueBuf); if (prev != Long.MIN_VALUE) { assert(prev <= key.get()); prev = key.get(); } LOG.info("key = " + key.get() + "; value = " + value); } reader.close(); }
public MapProcessor(){ super(true); }
MapProcessor extends MRTask implements LogicalIOProcessor { public MapProcessor(){ super(true); } }
MapProcessor extends MRTask implements LogicalIOProcessor { public MapProcessor(){ super(true); } MapProcessor(); }
MapProcessor extends MRTask implements LogicalIOProcessor { public MapProcessor(){ super(true); } MapProcessor(); @Override void initialize(TezProcessorContext processorContext); @Override void handleEvents(List<Event> processorEvents); void close(); @Override void run(Map<String, LogicalInput> inputs,
      Map<String, LogicalOutput> outputs); @Override void localizeConfiguration(JobConf jobConf); }
MapProcessor extends MRTask implements LogicalIOProcessor { public MapProcessor(){ super(true); } MapProcessor(); @Override void initialize(TezProcessorContext processorContext); @Override void handleEvents(List<Event> processorEvents); void close(); @Override void run(Map<String, LogicalInput> inputs,
      Map<String, LogicalOutput> outputs); @Override void localizeConfiguration(JobConf jobConf);  }