@Test public void testConvertDAGPlanToATSMap() throws IOException, JSONException { DAGPlan dagPlan = createDAG(); Map<String, Object> atsMap = DAGUtils.convertDAGPlanToATSMap(dagPlan); Assert.assertTrue(atsMap.containsKey(DAGUtils.DAG_NAME_KEY)); Assert.assertEquals(dagPlan.getName(), atsMap.get(DAGUtils.DAG_NAME_KEY)); Assert.assertTrue(atsMap.containsKey("version")); Assert.assertEquals(1, atsMap.get("version")); Assert.assertTrue(atsMap.containsKey(DAGUtils.VERTICES_KEY)); Assert.assertTrue(atsMap.containsKey(DAGUtils.EDGES_KEY)); Assert.assertTrue(atsMap.containsKey(DAGUtils.VERTEX_GROUPS_KEY)); Assert.assertEquals(3, ((Collection<?>) atsMap.get(DAGUtils.VERTICES_KEY)).size()); Set<String> vNames = Sets.newHashSet("vertex1", "vertex2", "vertex3"); Set<String> inEdgeIds = new HashSet<String>(); Set<String> outEdgeIds = new HashSet<String>(); int additionalInputCount = 0; int additionalOutputCount = 0; for (Object o : ((Collection<?>) atsMap.get(DAGUtils.VERTICES_KEY))) { Map<String, Object> v = (Map<String, Object>) o; Assert.assertTrue(v.containsKey(DAGUtils.VERTEX_NAME_KEY)); Assert.assertTrue(v.containsKey(DAGUtils.PROCESSOR_CLASS_KEY)); Assert.assertTrue(v.containsKey(DAGUtils.USER_PAYLOAD_AS_TEXT)); if (v.containsKey(DAGUtils.IN_EDGE_IDS_KEY)) { inEdgeIds.addAll(((Collection<String>) v.get(DAGUtils.IN_EDGE_IDS_KEY))); } if (v.containsKey(DAGUtils.OUT_EDGE_IDS_KEY)) { outEdgeIds.addAll(((Collection<String>) v.get(DAGUtils.OUT_EDGE_IDS_KEY))); } String vName = (String) v.get(DAGUtils.VERTEX_NAME_KEY); Assert.assertTrue(vNames.contains(vName)); String procPayload = vName + " Processor HistoryText"; Assert.assertEquals(procPayload, v.get(DAGUtils.USER_PAYLOAD_AS_TEXT)); if (v.containsKey(DAGUtils.ADDITIONAL_INPUTS_KEY)) { additionalInputCount += ((Collection<?>) v.get(DAGUtils.ADDITIONAL_INPUTS_KEY)).size(); for (Object input : ((Collection<?>) v.get(DAGUtils.ADDITIONAL_INPUTS_KEY))) { Map<String, Object> inputMap = (Map<String, Object>) input; Assert.assertTrue(inputMap.containsKey(DAGUtils.NAME_KEY)); Assert.assertTrue(inputMap.containsKey(DAGUtils.CLASS_KEY)); Assert.assertFalse(inputMap.containsKey(DAGUtils.INITIALIZER_KEY)); Assert.assertEquals("input HistoryText", inputMap.get(DAGUtils.USER_PAYLOAD_AS_TEXT)); } } if (v.containsKey(DAGUtils.ADDITIONAL_OUTPUTS_KEY)) { additionalOutputCount += ((Collection<?>) v.get(DAGUtils.ADDITIONAL_OUTPUTS_KEY)).size(); for (Object output : ((Collection<?>) v.get(DAGUtils.ADDITIONAL_OUTPUTS_KEY))) { Map<String, Object> outputMap = (Map<String, Object>) output; Assert.assertTrue(outputMap.containsKey(DAGUtils.NAME_KEY)); Assert.assertTrue(outputMap.containsKey(DAGUtils.CLASS_KEY)); Assert.assertTrue(outputMap.containsKey(DAGUtils.INITIALIZER_KEY)); Assert.assertEquals("uvOut HistoryText", outputMap.get(DAGUtils.USER_PAYLOAD_AS_TEXT)); } } } Assert.assertEquals(1, additionalInputCount); Assert.assertEquals(3, additionalOutputCount); Assert.assertEquals(2, inEdgeIds.size()); Assert.assertEquals(2, outEdgeIds.size()); for (Object o : ((Collection<?>) atsMap.get(DAGUtils.EDGES_KEY))) { Map<String, Object> e = (Map<String, Object>) o; Assert.assertTrue(inEdgeIds.contains(e.get(DAGUtils.EDGE_ID_KEY))); Assert.assertTrue(outEdgeIds.contains(e.get(DAGUtils.EDGE_ID_KEY))); Assert.assertTrue(e.containsKey(DAGUtils.INPUT_VERTEX_NAME_KEY)); Assert.assertTrue(e.containsKey(DAGUtils.OUTPUT_VERTEX_NAME_KEY)); Assert.assertEquals(DataMovementType.SCATTER_GATHER.name(), e.get(DAGUtils.DATA_MOVEMENT_TYPE_KEY)); Assert.assertEquals(DataSourceType.PERSISTED.name(), e.get(DAGUtils.DATA_SOURCE_TYPE_KEY)); Assert.assertEquals(SchedulingType.SEQUENTIAL.name(), e.get(DAGUtils.SCHEDULING_TYPE_KEY)); Assert.assertEquals("dummy output class", e.get(DAGUtils.EDGE_SOURCE_CLASS_KEY)); Assert.assertEquals("dummy input class", e.get(DAGUtils.EDGE_DESTINATION_CLASS_KEY)); Assert.assertEquals("Dummy History Text", e.get(DAGUtils.OUTPUT_USER_PAYLOAD_AS_TEXT)); Assert.assertEquals("Dummy History Text", e.get(DAGUtils.INPUT_USER_PAYLOAD_AS_TEXT)); } for (Object o : ((Collection<?>) atsMap.get(DAGUtils.VERTEX_GROUPS_KEY))) { Map<String, Object> e = (Map<String, Object>) o; Assert.assertEquals("uv12", e.get(DAGUtils.VERTEX_GROUP_NAME_KEY)); Assert.assertTrue(e.containsKey(DAGUtils.VERTEX_GROUP_MEMBERS_KEY)); Assert.assertTrue(e.containsKey(DAGUtils.VERTEX_GROUP_OUTPUTS_KEY)); Assert.assertTrue(e.containsKey(DAGUtils.VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY)); } }
public static Map<String,Object> convertDAGPlanToATSMap( DAGProtos.DAGPlan dagPlan) throws IOException { final String VERSION_KEY = "version"; final int version = 1; Map<String,Object> dagMap = new LinkedHashMap<String, Object>(); dagMap.put(DAG_NAME_KEY, dagPlan.getName()); dagMap.put(VERSION_KEY, version); ArrayList<Object> verticesList = new ArrayList<Object>(); for (DAGProtos.VertexPlan vertexPlan : dagPlan.getVertexList()) { Map<String,Object> vertexMap = new LinkedHashMap<String, Object>(); vertexMap.put(VERTEX_NAME_KEY, vertexPlan.getName()); if (vertexPlan.hasProcessorDescriptor()) { vertexMap.put(PROCESSOR_CLASS_KEY, vertexPlan.getProcessorDescriptor().getClassName()); if (vertexPlan.getProcessorDescriptor().hasHistoryText()) { vertexMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( vertexPlan.getProcessorDescriptor())); } } ArrayList<Object> inEdgeIdList = new ArrayList<Object>(); inEdgeIdList.addAll(vertexPlan.getInEdgeIdList()); putInto(vertexMap, IN_EDGE_IDS_KEY, inEdgeIdList); ArrayList<Object> outEdgeIdList = new ArrayList<Object>(); outEdgeIdList.addAll(vertexPlan.getOutEdgeIdList()); putInto(vertexMap, OUT_EDGE_IDS_KEY, outEdgeIdList); ArrayList<Object> inputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto input : vertexPlan.getInputsList()) { Map<String,Object> inputMap = new LinkedHashMap<String, Object>(); inputMap.put(NAME_KEY, input.getName()); inputMap.put(CLASS_KEY, input.getEntityDescriptor().getClassName()); if (input.hasInitializerClassName()) { inputMap.put(INITIALIZER_KEY, input.getInitializerClassName()); } if (input.getEntityDescriptor().hasHistoryText()) { inputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( input.getEntityDescriptor())); } inputsList.add(inputMap); } putInto(vertexMap, ADDITIONAL_INPUTS_KEY, inputsList); ArrayList<Object> outputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto output : vertexPlan.getOutputsList()) { Map<String,Object> outputMap = new LinkedHashMap<String, Object>(); outputMap.put(NAME_KEY, output.getName()); outputMap.put(CLASS_KEY, output.getEntityDescriptor().getClassName()); if (output.hasInitializerClassName()) { outputMap.put(INITIALIZER_KEY, output.getInitializerClassName()); } if (output.getEntityDescriptor().hasHistoryText()) { outputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( output.getEntityDescriptor())); } outputsList.add(outputMap); } putInto(vertexMap, ADDITIONAL_OUTPUTS_KEY, outputsList); if (vertexPlan.hasVertexManagerPlugin()) { vertexMap.put(VERTEX_MANAGER_PLUGIN_CLASS_KEY, vertexPlan.getVertexManagerPlugin().getClassName()); } verticesList.add(vertexMap); } putInto(dagMap, VERTICES_KEY, verticesList); ArrayList<Object> edgesList = new ArrayList<Object>(); for (DAGProtos.EdgePlan edgePlan : dagPlan.getEdgeList()) { Map<String,Object> edgeMap = new LinkedHashMap<String, Object>(); edgeMap.put(EDGE_ID_KEY, edgePlan.getId()); edgeMap.put(INPUT_VERTEX_NAME_KEY, edgePlan.getInputVertexName()); edgeMap.put(OUTPUT_VERTEX_NAME_KEY, edgePlan.getOutputVertexName()); edgeMap.put(DATA_MOVEMENT_TYPE_KEY, edgePlan.getDataMovementType().name()); edgeMap.put(DATA_SOURCE_TYPE_KEY, edgePlan.getDataSourceType().name()); edgeMap.put(SCHEDULING_TYPE_KEY, edgePlan.getSchedulingType().name()); edgeMap.put(EDGE_SOURCE_CLASS_KEY, edgePlan.getEdgeSource().getClassName()); edgeMap.put(EDGE_DESTINATION_CLASS_KEY, edgePlan.getEdgeDestination().getClassName()); if (edgePlan.getEdgeSource().hasHistoryText()) { edgeMap.put(OUTPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeSource())); } if (edgePlan.getEdgeDestination().hasHistoryText()) { edgeMap.put(INPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeDestination())); } edgesList.add(edgeMap); } putInto(dagMap, EDGES_KEY, edgesList); ArrayList<Object> vertexGroupsList = new ArrayList<Object>(); for (DAGProtos.PlanVertexGroupInfo vertexGroupInfo : dagPlan.getVertexGroupsList()) { Map<String,Object> groupMap = new LinkedHashMap<String, Object>(); groupMap.put(VERTEX_GROUP_NAME_KEY, vertexGroupInfo.getGroupName()); if (vertexGroupInfo.getGroupMembersCount() > 0 ) { groupMap.put(VERTEX_GROUP_MEMBERS_KEY, vertexGroupInfo.getGroupMembersList()); } if (vertexGroupInfo.getOutputsCount() > 0) { groupMap.put(VERTEX_GROUP_OUTPUTS_KEY, vertexGroupInfo.getOutputsList()); } if (vertexGroupInfo.getEdgeMergedInputsCount() > 0) { ArrayList<Object> edgeMergedInputs = new ArrayList<Object>(); for (PlanGroupInputEdgeInfo edgeMergedInputInfo : vertexGroupInfo.getEdgeMergedInputsList()) { Map<String,Object> edgeMergedInput = new LinkedHashMap<String, Object>(); edgeMergedInput.put(VERTEX_GROUP_DESTINATION_VERTEX_NAME_KEY, edgeMergedInputInfo.getDestVertexName()); if (edgeMergedInputInfo.hasMergedInput() && edgeMergedInputInfo.getMergedInput().hasClassName()) { edgeMergedInput.put(PROCESSOR_CLASS_KEY, edgeMergedInputInfo.getMergedInput().getClassName()); if (edgeMergedInputInfo.getMergedInput().hasHistoryText()) { edgeMergedInput.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgeMergedInputInfo.getMergedInput())); } } edgeMergedInputs.add(edgeMergedInput); } groupMap.put(VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY, edgeMergedInputs); } vertexGroupsList.add(groupMap); } putInto(dagMap, VERTEX_GROUPS_KEY, vertexGroupsList); return dagMap; }
DAGUtils { public static Map<String,Object> convertDAGPlanToATSMap( DAGProtos.DAGPlan dagPlan) throws IOException { final String VERSION_KEY = "version"; final int version = 1; Map<String,Object> dagMap = new LinkedHashMap<String, Object>(); dagMap.put(DAG_NAME_KEY, dagPlan.getName()); dagMap.put(VERSION_KEY, version); ArrayList<Object> verticesList = new ArrayList<Object>(); for (DAGProtos.VertexPlan vertexPlan : dagPlan.getVertexList()) { Map<String,Object> vertexMap = new LinkedHashMap<String, Object>(); vertexMap.put(VERTEX_NAME_KEY, vertexPlan.getName()); if (vertexPlan.hasProcessorDescriptor()) { vertexMap.put(PROCESSOR_CLASS_KEY, vertexPlan.getProcessorDescriptor().getClassName()); if (vertexPlan.getProcessorDescriptor().hasHistoryText()) { vertexMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( vertexPlan.getProcessorDescriptor())); } } ArrayList<Object> inEdgeIdList = new ArrayList<Object>(); inEdgeIdList.addAll(vertexPlan.getInEdgeIdList()); putInto(vertexMap, IN_EDGE_IDS_KEY, inEdgeIdList); ArrayList<Object> outEdgeIdList = new ArrayList<Object>(); outEdgeIdList.addAll(vertexPlan.getOutEdgeIdList()); putInto(vertexMap, OUT_EDGE_IDS_KEY, outEdgeIdList); ArrayList<Object> inputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto input : vertexPlan.getInputsList()) { Map<String,Object> inputMap = new LinkedHashMap<String, Object>(); inputMap.put(NAME_KEY, input.getName()); inputMap.put(CLASS_KEY, input.getEntityDescriptor().getClassName()); if (input.hasInitializerClassName()) { inputMap.put(INITIALIZER_KEY, input.getInitializerClassName()); } if (input.getEntityDescriptor().hasHistoryText()) { inputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( input.getEntityDescriptor())); } inputsList.add(inputMap); } putInto(vertexMap, ADDITIONAL_INPUTS_KEY, inputsList); ArrayList<Object> outputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto output : vertexPlan.getOutputsList()) { Map<String,Object> outputMap = new LinkedHashMap<String, Object>(); outputMap.put(NAME_KEY, output.getName()); outputMap.put(CLASS_KEY, output.getEntityDescriptor().getClassName()); if (output.hasInitializerClassName()) { outputMap.put(INITIALIZER_KEY, output.getInitializerClassName()); } if (output.getEntityDescriptor().hasHistoryText()) { outputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( output.getEntityDescriptor())); } outputsList.add(outputMap); } putInto(vertexMap, ADDITIONAL_OUTPUTS_KEY, outputsList); if (vertexPlan.hasVertexManagerPlugin()) { vertexMap.put(VERTEX_MANAGER_PLUGIN_CLASS_KEY, vertexPlan.getVertexManagerPlugin().getClassName()); } verticesList.add(vertexMap); } putInto(dagMap, VERTICES_KEY, verticesList); ArrayList<Object> edgesList = new ArrayList<Object>(); for (DAGProtos.EdgePlan edgePlan : dagPlan.getEdgeList()) { Map<String,Object> edgeMap = new LinkedHashMap<String, Object>(); edgeMap.put(EDGE_ID_KEY, edgePlan.getId()); edgeMap.put(INPUT_VERTEX_NAME_KEY, edgePlan.getInputVertexName()); edgeMap.put(OUTPUT_VERTEX_NAME_KEY, edgePlan.getOutputVertexName()); edgeMap.put(DATA_MOVEMENT_TYPE_KEY, edgePlan.getDataMovementType().name()); edgeMap.put(DATA_SOURCE_TYPE_KEY, edgePlan.getDataSourceType().name()); edgeMap.put(SCHEDULING_TYPE_KEY, edgePlan.getSchedulingType().name()); edgeMap.put(EDGE_SOURCE_CLASS_KEY, edgePlan.getEdgeSource().getClassName()); edgeMap.put(EDGE_DESTINATION_CLASS_KEY, edgePlan.getEdgeDestination().getClassName()); if (edgePlan.getEdgeSource().hasHistoryText()) { edgeMap.put(OUTPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeSource())); } if (edgePlan.getEdgeDestination().hasHistoryText()) { edgeMap.put(INPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeDestination())); } edgesList.add(edgeMap); } putInto(dagMap, EDGES_KEY, edgesList); ArrayList<Object> vertexGroupsList = new ArrayList<Object>(); for (DAGProtos.PlanVertexGroupInfo vertexGroupInfo : dagPlan.getVertexGroupsList()) { Map<String,Object> groupMap = new LinkedHashMap<String, Object>(); groupMap.put(VERTEX_GROUP_NAME_KEY, vertexGroupInfo.getGroupName()); if (vertexGroupInfo.getGroupMembersCount() > 0 ) { groupMap.put(VERTEX_GROUP_MEMBERS_KEY, vertexGroupInfo.getGroupMembersList()); } if (vertexGroupInfo.getOutputsCount() > 0) { groupMap.put(VERTEX_GROUP_OUTPUTS_KEY, vertexGroupInfo.getOutputsList()); } if (vertexGroupInfo.getEdgeMergedInputsCount() > 0) { ArrayList<Object> edgeMergedInputs = new ArrayList<Object>(); for (PlanGroupInputEdgeInfo edgeMergedInputInfo : vertexGroupInfo.getEdgeMergedInputsList()) { Map<String,Object> edgeMergedInput = new LinkedHashMap<String, Object>(); edgeMergedInput.put(VERTEX_GROUP_DESTINATION_VERTEX_NAME_KEY, edgeMergedInputInfo.getDestVertexName()); if (edgeMergedInputInfo.hasMergedInput() && edgeMergedInputInfo.getMergedInput().hasClassName()) { edgeMergedInput.put(PROCESSOR_CLASS_KEY, edgeMergedInputInfo.getMergedInput().getClassName()); if (edgeMergedInputInfo.getMergedInput().hasHistoryText()) { edgeMergedInput.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgeMergedInputInfo.getMergedInput())); } } edgeMergedInputs.add(edgeMergedInput); } groupMap.put(VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY, edgeMergedInputs); } vertexGroupsList.add(groupMap); } putInto(dagMap, VERTEX_GROUPS_KEY, vertexGroupsList); return dagMap; } }
DAGUtils { public static Map<String,Object> convertDAGPlanToATSMap( DAGProtos.DAGPlan dagPlan) throws IOException { final String VERSION_KEY = "version"; final int version = 1; Map<String,Object> dagMap = new LinkedHashMap<String, Object>(); dagMap.put(DAG_NAME_KEY, dagPlan.getName()); dagMap.put(VERSION_KEY, version); ArrayList<Object> verticesList = new ArrayList<Object>(); for (DAGProtos.VertexPlan vertexPlan : dagPlan.getVertexList()) { Map<String,Object> vertexMap = new LinkedHashMap<String, Object>(); vertexMap.put(VERTEX_NAME_KEY, vertexPlan.getName()); if (vertexPlan.hasProcessorDescriptor()) { vertexMap.put(PROCESSOR_CLASS_KEY, vertexPlan.getProcessorDescriptor().getClassName()); if (vertexPlan.getProcessorDescriptor().hasHistoryText()) { vertexMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( vertexPlan.getProcessorDescriptor())); } } ArrayList<Object> inEdgeIdList = new ArrayList<Object>(); inEdgeIdList.addAll(vertexPlan.getInEdgeIdList()); putInto(vertexMap, IN_EDGE_IDS_KEY, inEdgeIdList); ArrayList<Object> outEdgeIdList = new ArrayList<Object>(); outEdgeIdList.addAll(vertexPlan.getOutEdgeIdList()); putInto(vertexMap, OUT_EDGE_IDS_KEY, outEdgeIdList); ArrayList<Object> inputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto input : vertexPlan.getInputsList()) { Map<String,Object> inputMap = new LinkedHashMap<String, Object>(); inputMap.put(NAME_KEY, input.getName()); inputMap.put(CLASS_KEY, input.getEntityDescriptor().getClassName()); if (input.hasInitializerClassName()) { inputMap.put(INITIALIZER_KEY, input.getInitializerClassName()); } if (input.getEntityDescriptor().hasHistoryText()) { inputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( input.getEntityDescriptor())); } inputsList.add(inputMap); } putInto(vertexMap, ADDITIONAL_INPUTS_KEY, inputsList); ArrayList<Object> outputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto output : vertexPlan.getOutputsList()) { Map<String,Object> outputMap = new LinkedHashMap<String, Object>(); outputMap.put(NAME_KEY, output.getName()); outputMap.put(CLASS_KEY, output.getEntityDescriptor().getClassName()); if (output.hasInitializerClassName()) { outputMap.put(INITIALIZER_KEY, output.getInitializerClassName()); } if (output.getEntityDescriptor().hasHistoryText()) { outputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( output.getEntityDescriptor())); } outputsList.add(outputMap); } putInto(vertexMap, ADDITIONAL_OUTPUTS_KEY, outputsList); if (vertexPlan.hasVertexManagerPlugin()) { vertexMap.put(VERTEX_MANAGER_PLUGIN_CLASS_KEY, vertexPlan.getVertexManagerPlugin().getClassName()); } verticesList.add(vertexMap); } putInto(dagMap, VERTICES_KEY, verticesList); ArrayList<Object> edgesList = new ArrayList<Object>(); for (DAGProtos.EdgePlan edgePlan : dagPlan.getEdgeList()) { Map<String,Object> edgeMap = new LinkedHashMap<String, Object>(); edgeMap.put(EDGE_ID_KEY, edgePlan.getId()); edgeMap.put(INPUT_VERTEX_NAME_KEY, edgePlan.getInputVertexName()); edgeMap.put(OUTPUT_VERTEX_NAME_KEY, edgePlan.getOutputVertexName()); edgeMap.put(DATA_MOVEMENT_TYPE_KEY, edgePlan.getDataMovementType().name()); edgeMap.put(DATA_SOURCE_TYPE_KEY, edgePlan.getDataSourceType().name()); edgeMap.put(SCHEDULING_TYPE_KEY, edgePlan.getSchedulingType().name()); edgeMap.put(EDGE_SOURCE_CLASS_KEY, edgePlan.getEdgeSource().getClassName()); edgeMap.put(EDGE_DESTINATION_CLASS_KEY, edgePlan.getEdgeDestination().getClassName()); if (edgePlan.getEdgeSource().hasHistoryText()) { edgeMap.put(OUTPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeSource())); } if (edgePlan.getEdgeDestination().hasHistoryText()) { edgeMap.put(INPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeDestination())); } edgesList.add(edgeMap); } putInto(dagMap, EDGES_KEY, edgesList); ArrayList<Object> vertexGroupsList = new ArrayList<Object>(); for (DAGProtos.PlanVertexGroupInfo vertexGroupInfo : dagPlan.getVertexGroupsList()) { Map<String,Object> groupMap = new LinkedHashMap<String, Object>(); groupMap.put(VERTEX_GROUP_NAME_KEY, vertexGroupInfo.getGroupName()); if (vertexGroupInfo.getGroupMembersCount() > 0 ) { groupMap.put(VERTEX_GROUP_MEMBERS_KEY, vertexGroupInfo.getGroupMembersList()); } if (vertexGroupInfo.getOutputsCount() > 0) { groupMap.put(VERTEX_GROUP_OUTPUTS_KEY, vertexGroupInfo.getOutputsList()); } if (vertexGroupInfo.getEdgeMergedInputsCount() > 0) { ArrayList<Object> edgeMergedInputs = new ArrayList<Object>(); for (PlanGroupInputEdgeInfo edgeMergedInputInfo : vertexGroupInfo.getEdgeMergedInputsList()) { Map<String,Object> edgeMergedInput = new LinkedHashMap<String, Object>(); edgeMergedInput.put(VERTEX_GROUP_DESTINATION_VERTEX_NAME_KEY, edgeMergedInputInfo.getDestVertexName()); if (edgeMergedInputInfo.hasMergedInput() && edgeMergedInputInfo.getMergedInput().hasClassName()) { edgeMergedInput.put(PROCESSOR_CLASS_KEY, edgeMergedInputInfo.getMergedInput().getClassName()); if (edgeMergedInputInfo.getMergedInput().hasHistoryText()) { edgeMergedInput.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgeMergedInputInfo.getMergedInput())); } } edgeMergedInputs.add(edgeMergedInput); } groupMap.put(VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY, edgeMergedInputs); } vertexGroupsList.add(groupMap); } putInto(dagMap, VERTEX_GROUPS_KEY, vertexGroupsList); return dagMap; }  }
DAGUtils { public static Map<String,Object> convertDAGPlanToATSMap( DAGProtos.DAGPlan dagPlan) throws IOException { final String VERSION_KEY = "version"; final int version = 1; Map<String,Object> dagMap = new LinkedHashMap<String, Object>(); dagMap.put(DAG_NAME_KEY, dagPlan.getName()); dagMap.put(VERSION_KEY, version); ArrayList<Object> verticesList = new ArrayList<Object>(); for (DAGProtos.VertexPlan vertexPlan : dagPlan.getVertexList()) { Map<String,Object> vertexMap = new LinkedHashMap<String, Object>(); vertexMap.put(VERTEX_NAME_KEY, vertexPlan.getName()); if (vertexPlan.hasProcessorDescriptor()) { vertexMap.put(PROCESSOR_CLASS_KEY, vertexPlan.getProcessorDescriptor().getClassName()); if (vertexPlan.getProcessorDescriptor().hasHistoryText()) { vertexMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( vertexPlan.getProcessorDescriptor())); } } ArrayList<Object> inEdgeIdList = new ArrayList<Object>(); inEdgeIdList.addAll(vertexPlan.getInEdgeIdList()); putInto(vertexMap, IN_EDGE_IDS_KEY, inEdgeIdList); ArrayList<Object> outEdgeIdList = new ArrayList<Object>(); outEdgeIdList.addAll(vertexPlan.getOutEdgeIdList()); putInto(vertexMap, OUT_EDGE_IDS_KEY, outEdgeIdList); ArrayList<Object> inputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto input : vertexPlan.getInputsList()) { Map<String,Object> inputMap = new LinkedHashMap<String, Object>(); inputMap.put(NAME_KEY, input.getName()); inputMap.put(CLASS_KEY, input.getEntityDescriptor().getClassName()); if (input.hasInitializerClassName()) { inputMap.put(INITIALIZER_KEY, input.getInitializerClassName()); } if (input.getEntityDescriptor().hasHistoryText()) { inputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( input.getEntityDescriptor())); } inputsList.add(inputMap); } putInto(vertexMap, ADDITIONAL_INPUTS_KEY, inputsList); ArrayList<Object> outputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto output : vertexPlan.getOutputsList()) { Map<String,Object> outputMap = new LinkedHashMap<String, Object>(); outputMap.put(NAME_KEY, output.getName()); outputMap.put(CLASS_KEY, output.getEntityDescriptor().getClassName()); if (output.hasInitializerClassName()) { outputMap.put(INITIALIZER_KEY, output.getInitializerClassName()); } if (output.getEntityDescriptor().hasHistoryText()) { outputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( output.getEntityDescriptor())); } outputsList.add(outputMap); } putInto(vertexMap, ADDITIONAL_OUTPUTS_KEY, outputsList); if (vertexPlan.hasVertexManagerPlugin()) { vertexMap.put(VERTEX_MANAGER_PLUGIN_CLASS_KEY, vertexPlan.getVertexManagerPlugin().getClassName()); } verticesList.add(vertexMap); } putInto(dagMap, VERTICES_KEY, verticesList); ArrayList<Object> edgesList = new ArrayList<Object>(); for (DAGProtos.EdgePlan edgePlan : dagPlan.getEdgeList()) { Map<String,Object> edgeMap = new LinkedHashMap<String, Object>(); edgeMap.put(EDGE_ID_KEY, edgePlan.getId()); edgeMap.put(INPUT_VERTEX_NAME_KEY, edgePlan.getInputVertexName()); edgeMap.put(OUTPUT_VERTEX_NAME_KEY, edgePlan.getOutputVertexName()); edgeMap.put(DATA_MOVEMENT_TYPE_KEY, edgePlan.getDataMovementType().name()); edgeMap.put(DATA_SOURCE_TYPE_KEY, edgePlan.getDataSourceType().name()); edgeMap.put(SCHEDULING_TYPE_KEY, edgePlan.getSchedulingType().name()); edgeMap.put(EDGE_SOURCE_CLASS_KEY, edgePlan.getEdgeSource().getClassName()); edgeMap.put(EDGE_DESTINATION_CLASS_KEY, edgePlan.getEdgeDestination().getClassName()); if (edgePlan.getEdgeSource().hasHistoryText()) { edgeMap.put(OUTPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeSource())); } if (edgePlan.getEdgeDestination().hasHistoryText()) { edgeMap.put(INPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeDestination())); } edgesList.add(edgeMap); } putInto(dagMap, EDGES_KEY, edgesList); ArrayList<Object> vertexGroupsList = new ArrayList<Object>(); for (DAGProtos.PlanVertexGroupInfo vertexGroupInfo : dagPlan.getVertexGroupsList()) { Map<String,Object> groupMap = new LinkedHashMap<String, Object>(); groupMap.put(VERTEX_GROUP_NAME_KEY, vertexGroupInfo.getGroupName()); if (vertexGroupInfo.getGroupMembersCount() > 0 ) { groupMap.put(VERTEX_GROUP_MEMBERS_KEY, vertexGroupInfo.getGroupMembersList()); } if (vertexGroupInfo.getOutputsCount() > 0) { groupMap.put(VERTEX_GROUP_OUTPUTS_KEY, vertexGroupInfo.getOutputsList()); } if (vertexGroupInfo.getEdgeMergedInputsCount() > 0) { ArrayList<Object> edgeMergedInputs = new ArrayList<Object>(); for (PlanGroupInputEdgeInfo edgeMergedInputInfo : vertexGroupInfo.getEdgeMergedInputsList()) { Map<String,Object> edgeMergedInput = new LinkedHashMap<String, Object>(); edgeMergedInput.put(VERTEX_GROUP_DESTINATION_VERTEX_NAME_KEY, edgeMergedInputInfo.getDestVertexName()); if (edgeMergedInputInfo.hasMergedInput() && edgeMergedInputInfo.getMergedInput().hasClassName()) { edgeMergedInput.put(PROCESSOR_CLASS_KEY, edgeMergedInputInfo.getMergedInput().getClassName()); if (edgeMergedInputInfo.getMergedInput().hasHistoryText()) { edgeMergedInput.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgeMergedInputInfo.getMergedInput())); } } edgeMergedInputs.add(edgeMergedInput); } groupMap.put(VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY, edgeMergedInputs); } vertexGroupsList.add(groupMap); } putInto(dagMap, VERTEX_GROUPS_KEY, vertexGroupsList); return dagMap; }  static JSONObject generateSimpleJSONPlan(DAGProtos.DAGPlan dagPlan); static JSONObject convertCountersToJSON(TezCounters counters); static Map<String,Object> convertCountersToATSMap(TezCounters counters); static Map<String,Object> convertDAGPlanToATSMap(
      DAGProtos.DAGPlan dagPlan); static Map<String,Object> convertVertexStatsToATSMap(
      VertexStats vertexStats); }
DAGUtils { public static Map<String,Object> convertDAGPlanToATSMap( DAGProtos.DAGPlan dagPlan) throws IOException { final String VERSION_KEY = "version"; final int version = 1; Map<String,Object> dagMap = new LinkedHashMap<String, Object>(); dagMap.put(DAG_NAME_KEY, dagPlan.getName()); dagMap.put(VERSION_KEY, version); ArrayList<Object> verticesList = new ArrayList<Object>(); for (DAGProtos.VertexPlan vertexPlan : dagPlan.getVertexList()) { Map<String,Object> vertexMap = new LinkedHashMap<String, Object>(); vertexMap.put(VERTEX_NAME_KEY, vertexPlan.getName()); if (vertexPlan.hasProcessorDescriptor()) { vertexMap.put(PROCESSOR_CLASS_KEY, vertexPlan.getProcessorDescriptor().getClassName()); if (vertexPlan.getProcessorDescriptor().hasHistoryText()) { vertexMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( vertexPlan.getProcessorDescriptor())); } } ArrayList<Object> inEdgeIdList = new ArrayList<Object>(); inEdgeIdList.addAll(vertexPlan.getInEdgeIdList()); putInto(vertexMap, IN_EDGE_IDS_KEY, inEdgeIdList); ArrayList<Object> outEdgeIdList = new ArrayList<Object>(); outEdgeIdList.addAll(vertexPlan.getOutEdgeIdList()); putInto(vertexMap, OUT_EDGE_IDS_KEY, outEdgeIdList); ArrayList<Object> inputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto input : vertexPlan.getInputsList()) { Map<String,Object> inputMap = new LinkedHashMap<String, Object>(); inputMap.put(NAME_KEY, input.getName()); inputMap.put(CLASS_KEY, input.getEntityDescriptor().getClassName()); if (input.hasInitializerClassName()) { inputMap.put(INITIALIZER_KEY, input.getInitializerClassName()); } if (input.getEntityDescriptor().hasHistoryText()) { inputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( input.getEntityDescriptor())); } inputsList.add(inputMap); } putInto(vertexMap, ADDITIONAL_INPUTS_KEY, inputsList); ArrayList<Object> outputsList = new ArrayList<Object>(); for (DAGProtos.RootInputLeafOutputProto output : vertexPlan.getOutputsList()) { Map<String,Object> outputMap = new LinkedHashMap<String, Object>(); outputMap.put(NAME_KEY, output.getName()); outputMap.put(CLASS_KEY, output.getEntityDescriptor().getClassName()); if (output.hasInitializerClassName()) { outputMap.put(INITIALIZER_KEY, output.getInitializerClassName()); } if (output.getEntityDescriptor().hasHistoryText()) { outputMap.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( output.getEntityDescriptor())); } outputsList.add(outputMap); } putInto(vertexMap, ADDITIONAL_OUTPUTS_KEY, outputsList); if (vertexPlan.hasVertexManagerPlugin()) { vertexMap.put(VERTEX_MANAGER_PLUGIN_CLASS_KEY, vertexPlan.getVertexManagerPlugin().getClassName()); } verticesList.add(vertexMap); } putInto(dagMap, VERTICES_KEY, verticesList); ArrayList<Object> edgesList = new ArrayList<Object>(); for (DAGProtos.EdgePlan edgePlan : dagPlan.getEdgeList()) { Map<String,Object> edgeMap = new LinkedHashMap<String, Object>(); edgeMap.put(EDGE_ID_KEY, edgePlan.getId()); edgeMap.put(INPUT_VERTEX_NAME_KEY, edgePlan.getInputVertexName()); edgeMap.put(OUTPUT_VERTEX_NAME_KEY, edgePlan.getOutputVertexName()); edgeMap.put(DATA_MOVEMENT_TYPE_KEY, edgePlan.getDataMovementType().name()); edgeMap.put(DATA_SOURCE_TYPE_KEY, edgePlan.getDataSourceType().name()); edgeMap.put(SCHEDULING_TYPE_KEY, edgePlan.getSchedulingType().name()); edgeMap.put(EDGE_SOURCE_CLASS_KEY, edgePlan.getEdgeSource().getClassName()); edgeMap.put(EDGE_DESTINATION_CLASS_KEY, edgePlan.getEdgeDestination().getClassName()); if (edgePlan.getEdgeSource().hasHistoryText()) { edgeMap.put(OUTPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeSource())); } if (edgePlan.getEdgeDestination().hasHistoryText()) { edgeMap.put(INPUT_USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgePlan.getEdgeDestination())); } edgesList.add(edgeMap); } putInto(dagMap, EDGES_KEY, edgesList); ArrayList<Object> vertexGroupsList = new ArrayList<Object>(); for (DAGProtos.PlanVertexGroupInfo vertexGroupInfo : dagPlan.getVertexGroupsList()) { Map<String,Object> groupMap = new LinkedHashMap<String, Object>(); groupMap.put(VERTEX_GROUP_NAME_KEY, vertexGroupInfo.getGroupName()); if (vertexGroupInfo.getGroupMembersCount() > 0 ) { groupMap.put(VERTEX_GROUP_MEMBERS_KEY, vertexGroupInfo.getGroupMembersList()); } if (vertexGroupInfo.getOutputsCount() > 0) { groupMap.put(VERTEX_GROUP_OUTPUTS_KEY, vertexGroupInfo.getOutputsList()); } if (vertexGroupInfo.getEdgeMergedInputsCount() > 0) { ArrayList<Object> edgeMergedInputs = new ArrayList<Object>(); for (PlanGroupInputEdgeInfo edgeMergedInputInfo : vertexGroupInfo.getEdgeMergedInputsList()) { Map<String,Object> edgeMergedInput = new LinkedHashMap<String, Object>(); edgeMergedInput.put(VERTEX_GROUP_DESTINATION_VERTEX_NAME_KEY, edgeMergedInputInfo.getDestVertexName()); if (edgeMergedInputInfo.hasMergedInput() && edgeMergedInputInfo.getMergedInput().hasClassName()) { edgeMergedInput.put(PROCESSOR_CLASS_KEY, edgeMergedInputInfo.getMergedInput().getClassName()); if (edgeMergedInputInfo.getMergedInput().hasHistoryText()) { edgeMergedInput.put(USER_PAYLOAD_AS_TEXT, DagTypeConverters.getHistoryTextFromProto( edgeMergedInputInfo.getMergedInput())); } } edgeMergedInputs.add(edgeMergedInput); } groupMap.put(VERTEX_GROUP_EDGE_MERGED_INPUTS_KEY, edgeMergedInputs); } vertexGroupsList.add(groupMap); } putInto(dagMap, VERTEX_GROUPS_KEY, vertexGroupsList); return dagMap; }  static JSONObject generateSimpleJSONPlan(DAGProtos.DAGPlan dagPlan); static JSONObject convertCountersToJSON(TezCounters counters); static Map<String,Object> convertCountersToATSMap(TezCounters counters); static Map<String,Object> convertDAGPlanToATSMap(
      DAGProtos.DAGPlan dagPlan); static Map<String,Object> convertVertexStatsToATSMap(
      VertexStats vertexStats);  }