@Test public void testDeserializedPayload() throws IOException { Configuration conf = new Configuration(false); conf.setBoolean(MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, false); ByteString confByteString = MRHelpers.createByteStringFromConf(conf); InputSplit split1 = new InputSplitForTest(1); InputSplit split2 = new InputSplitForTest(2); MRSplitProto proto1 = MRHelpers.createSplitProto(split1); MRSplitProto proto2 = MRHelpers.createSplitProto(split2); MRSplitsProto.Builder splitsProtoBuilder = MRSplitsProto.newBuilder(); splitsProtoBuilder.addSplits(proto1); splitsProtoBuilder.addSplits(proto2); MRInputUserPayloadProto.Builder payloadProto = MRInputUserPayloadProto.newBuilder(); payloadProto.setSplits(splitsProtoBuilder.build()); payloadProto.setConfigurationBytes(confByteString); byte[] userPayload = payloadProto.build().toByteArray(); TezRootInputInitializerContext context = new TezRootInputInitializerContextForTest(userPayload); MRInputSplitDistributor splitDist = new MRInputSplitDistributor(); List<Event> events = splitDist.initialize(context); assertEquals(3, events.size()); assertTrue(events.get(0) instanceof RootInputUpdatePayloadEvent); assertTrue(events.get(1) instanceof RootInputDataInformationEvent); assertTrue(events.get(2) instanceof RootInputDataInformationEvent); RootInputDataInformationEvent diEvent1 = (RootInputDataInformationEvent) (events.get(1)); RootInputDataInformationEvent diEvent2 = (RootInputDataInformationEvent) (events.get(2)); assertNull(diEvent1.getUserPayload()); assertNull(diEvent2.getUserPayload()); assertNotNull(diEvent1.getDeserializedUserPayload()); assertNotNull(diEvent2.getDeserializedUserPayload()); assertTrue(diEvent1.getDeserializedUserPayload() instanceof InputSplitForTest); assertEquals(1, ((InputSplitForTest) diEvent1.getDeserializedUserPayload()).identifier); assertTrue(diEvent2.getDeserializedUserPayload() instanceof InputSplitForTest); assertEquals(2, ((InputSplitForTest) diEvent2.getDeserializedUserPayload()).identifier); }
@Override public List<Event> initialize(TezRootInputInitializerContext rootInputContext) throws IOException { Stopwatch sw = null; if (LOG.isDebugEnabled()) { sw = new Stopwatch().start(); } MRInputUserPayloadProto userPayloadProto = MRHelpers.parseMRInputPayload(rootInputContext.getUserPayload()); if (LOG.isDebugEnabled()) { sw.stop(); LOG.debug("Time to parse MRInput payload into prot: " + sw.elapsedMillis()); } Configuration conf = MRHelpers.createConfFromByteString(userPayloadProto .getConfigurationBytes()); JobConf jobConf = new JobConf(conf); boolean useNewApi = jobConf.getUseNewMapper(); sendSerializedEvents = conf.getBoolean( MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD_DEFAULT); LOG.info("Emitting serialized splits: " + sendSerializedEvents); this.splitsProto = userPayloadProto.getSplits(); MRInputUserPayloadProto.Builder updatedPayloadBuilder = MRInputUserPayloadProto.newBuilder(userPayloadProto); updatedPayloadBuilder.clearSplits(); List<Event> events = Lists.newArrayListWithCapacity(this.splitsProto.getSplitsCount() + 1); RootInputUpdatePayloadEvent updatePayloadEvent = new RootInputUpdatePayloadEvent( updatedPayloadBuilder.build().toByteArray()); events.add(updatePayloadEvent); int count = 0; for (MRSplitProto mrSplit : this.splitsProto.getSplitsList()) { RootInputDataInformationEvent diEvent; if (sendSerializedEvents) { diEvent = new RootInputDataInformationEvent(count++, mrSplit.toByteArray()); } else { if (useNewApi) { org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils .getNewSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, newInputSplit); } else { org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils .getOldSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, oldInputSplit); } } events.add(diEvent); } return events; }
MRInputSplitDistributor implements TezRootInputInitializer { @Override public List<Event> initialize(TezRootInputInitializerContext rootInputContext) throws IOException { Stopwatch sw = null; if (LOG.isDebugEnabled()) { sw = new Stopwatch().start(); } MRInputUserPayloadProto userPayloadProto = MRHelpers.parseMRInputPayload(rootInputContext.getUserPayload()); if (LOG.isDebugEnabled()) { sw.stop(); LOG.debug("Time to parse MRInput payload into prot: " + sw.elapsedMillis()); } Configuration conf = MRHelpers.createConfFromByteString(userPayloadProto .getConfigurationBytes()); JobConf jobConf = new JobConf(conf); boolean useNewApi = jobConf.getUseNewMapper(); sendSerializedEvents = conf.getBoolean( MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD_DEFAULT); LOG.info("Emitting serialized splits: " + sendSerializedEvents); this.splitsProto = userPayloadProto.getSplits(); MRInputUserPayloadProto.Builder updatedPayloadBuilder = MRInputUserPayloadProto.newBuilder(userPayloadProto); updatedPayloadBuilder.clearSplits(); List<Event> events = Lists.newArrayListWithCapacity(this.splitsProto.getSplitsCount() + 1); RootInputUpdatePayloadEvent updatePayloadEvent = new RootInputUpdatePayloadEvent( updatedPayloadBuilder.build().toByteArray()); events.add(updatePayloadEvent); int count = 0; for (MRSplitProto mrSplit : this.splitsProto.getSplitsList()) { RootInputDataInformationEvent diEvent; if (sendSerializedEvents) { diEvent = new RootInputDataInformationEvent(count++, mrSplit.toByteArray()); } else { if (useNewApi) { org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils .getNewSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, newInputSplit); } else { org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils .getOldSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, oldInputSplit); } } events.add(diEvent); } return events; } }
MRInputSplitDistributor implements TezRootInputInitializer { @Override public List<Event> initialize(TezRootInputInitializerContext rootInputContext) throws IOException { Stopwatch sw = null; if (LOG.isDebugEnabled()) { sw = new Stopwatch().start(); } MRInputUserPayloadProto userPayloadProto = MRHelpers.parseMRInputPayload(rootInputContext.getUserPayload()); if (LOG.isDebugEnabled()) { sw.stop(); LOG.debug("Time to parse MRInput payload into prot: " + sw.elapsedMillis()); } Configuration conf = MRHelpers.createConfFromByteString(userPayloadProto .getConfigurationBytes()); JobConf jobConf = new JobConf(conf); boolean useNewApi = jobConf.getUseNewMapper(); sendSerializedEvents = conf.getBoolean( MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD_DEFAULT); LOG.info("Emitting serialized splits: " + sendSerializedEvents); this.splitsProto = userPayloadProto.getSplits(); MRInputUserPayloadProto.Builder updatedPayloadBuilder = MRInputUserPayloadProto.newBuilder(userPayloadProto); updatedPayloadBuilder.clearSplits(); List<Event> events = Lists.newArrayListWithCapacity(this.splitsProto.getSplitsCount() + 1); RootInputUpdatePayloadEvent updatePayloadEvent = new RootInputUpdatePayloadEvent( updatedPayloadBuilder.build().toByteArray()); events.add(updatePayloadEvent); int count = 0; for (MRSplitProto mrSplit : this.splitsProto.getSplitsList()) { RootInputDataInformationEvent diEvent; if (sendSerializedEvents) { diEvent = new RootInputDataInformationEvent(count++, mrSplit.toByteArray()); } else { if (useNewApi) { org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils .getNewSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, newInputSplit); } else { org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils .getOldSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, oldInputSplit); } } events.add(diEvent); } return events; } MRInputSplitDistributor(); }
MRInputSplitDistributor implements TezRootInputInitializer { @Override public List<Event> initialize(TezRootInputInitializerContext rootInputContext) throws IOException { Stopwatch sw = null; if (LOG.isDebugEnabled()) { sw = new Stopwatch().start(); } MRInputUserPayloadProto userPayloadProto = MRHelpers.parseMRInputPayload(rootInputContext.getUserPayload()); if (LOG.isDebugEnabled()) { sw.stop(); LOG.debug("Time to parse MRInput payload into prot: " + sw.elapsedMillis()); } Configuration conf = MRHelpers.createConfFromByteString(userPayloadProto .getConfigurationBytes()); JobConf jobConf = new JobConf(conf); boolean useNewApi = jobConf.getUseNewMapper(); sendSerializedEvents = conf.getBoolean( MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD_DEFAULT); LOG.info("Emitting serialized splits: " + sendSerializedEvents); this.splitsProto = userPayloadProto.getSplits(); MRInputUserPayloadProto.Builder updatedPayloadBuilder = MRInputUserPayloadProto.newBuilder(userPayloadProto); updatedPayloadBuilder.clearSplits(); List<Event> events = Lists.newArrayListWithCapacity(this.splitsProto.getSplitsCount() + 1); RootInputUpdatePayloadEvent updatePayloadEvent = new RootInputUpdatePayloadEvent( updatedPayloadBuilder.build().toByteArray()); events.add(updatePayloadEvent); int count = 0; for (MRSplitProto mrSplit : this.splitsProto.getSplitsList()) { RootInputDataInformationEvent diEvent; if (sendSerializedEvents) { diEvent = new RootInputDataInformationEvent(count++, mrSplit.toByteArray()); } else { if (useNewApi) { org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils .getNewSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, newInputSplit); } else { org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils .getOldSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, oldInputSplit); } } events.add(diEvent); } return events; } MRInputSplitDistributor(); @Override List<Event> initialize(TezRootInputInitializerContext rootInputContext); @Override void handleInputInitializerEvent(List<RootInputInitializerEvent> events); }
MRInputSplitDistributor implements TezRootInputInitializer { @Override public List<Event> initialize(TezRootInputInitializerContext rootInputContext) throws IOException { Stopwatch sw = null; if (LOG.isDebugEnabled()) { sw = new Stopwatch().start(); } MRInputUserPayloadProto userPayloadProto = MRHelpers.parseMRInputPayload(rootInputContext.getUserPayload()); if (LOG.isDebugEnabled()) { sw.stop(); LOG.debug("Time to parse MRInput payload into prot: " + sw.elapsedMillis()); } Configuration conf = MRHelpers.createConfFromByteString(userPayloadProto .getConfigurationBytes()); JobConf jobConf = new JobConf(conf); boolean useNewApi = jobConf.getUseNewMapper(); sendSerializedEvents = conf.getBoolean( MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD, MRJobConfig.MR_TEZ_INPUT_INITIALIZER_SERIALIZE_EVENT_PAYLOAD_DEFAULT); LOG.info("Emitting serialized splits: " + sendSerializedEvents); this.splitsProto = userPayloadProto.getSplits(); MRInputUserPayloadProto.Builder updatedPayloadBuilder = MRInputUserPayloadProto.newBuilder(userPayloadProto); updatedPayloadBuilder.clearSplits(); List<Event> events = Lists.newArrayListWithCapacity(this.splitsProto.getSplitsCount() + 1); RootInputUpdatePayloadEvent updatePayloadEvent = new RootInputUpdatePayloadEvent( updatedPayloadBuilder.build().toByteArray()); events.add(updatePayloadEvent); int count = 0; for (MRSplitProto mrSplit : this.splitsProto.getSplitsList()) { RootInputDataInformationEvent diEvent; if (sendSerializedEvents) { diEvent = new RootInputDataInformationEvent(count++, mrSplit.toByteArray()); } else { if (useNewApi) { org.apache.hadoop.mapreduce.InputSplit newInputSplit = MRInputUtils .getNewSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, newInputSplit); } else { org.apache.hadoop.mapred.InputSplit oldInputSplit = MRInputUtils .getOldSplitDetailsFromEvent(mrSplit, conf); diEvent = new RootInputDataInformationEvent(count++, oldInputSplit); } } events.add(diEvent); } return events; } MRInputSplitDistributor(); @Override List<Event> initialize(TezRootInputInitializerContext rootInputContext); @Override void handleInputInitializerEvent(List<RootInputInitializerEvent> events);  }