@Test public void testPipelineGenTree() { MetaExplorer metaExplorer = new MockMetaExplorer(); JobConf job = new BridgeJobConf(); Pipeline pipelineOld = Pipeline.builder() .addMapper(WordCount.TokenizerMapper.class) .setOutputKeySchema(new Column[]{new Column("word", OdpsType.STRING)}) .setOutputValueSchema(new Column[]{new Column("count", OdpsType.BIGINT)}) .addReducer(WordCount.SumReducer.class, new Column[]{new Column("word", OdpsType.STRING)}, new Column[]{new Column("count", OdpsType.BIGINT)}, null, null, null, null, null) .addReducer(WordCount.SumReducer.class) .createPipeline(); Pipeline.toJobConf(job, pipelineOld); TableInfo tableIn = TableInfo.builder().projectName("foo").tableName("mr_src").build(); InputUtils.addTable(tableIn, job); TableInfo tableOut = TableInfo.builder().projectName("foo").tableName("mr_dst").build(); OutputUtils.addTable(tableOut, job); ((BridgeJobConf) job).setInputSchema(tableIn, SchemaUtils.fromString("word:string")); ((BridgeJobConf) job).setOutputSchema(SchemaUtils.fromString("word:string,count:bigint"), TableInfo.DEFAULT_LABEL); Pipeline pipeline = Pipeline.fromJobConf(job); assertNotNull(pipeline); LOTGenerator lotGen = new LOTGenerator(metaExplorer.getDefaultProject(), job, pipeline); LogicalOperatorTree lot = lotGen.genTree(); Assert.assertEquals(10L, lot.getOperatorsCount()); Assert.assertEquals("DataSource_0", lot.getOperators(0).getDataSource().getId()); Assert.assertEquals("MapTransform_1", lot.getOperators(1).getTransform().getId()); Assert.assertEquals("DIS_2", lot.getOperators(2).getDistributeBy().getId()); Assert.assertEquals("SORT_3", lot.getOperators(3).getSortBy().getId()); Assert.assertEquals("ReduceTransform_4", lot.getOperators(4).getTransform().getId()); Assert.assertEquals("DIS_5", lot.getOperators(5).getDistributeBy().getId()); Assert.assertEquals("SORT_6", lot.getOperators(6).getSortBy().getId()); Assert.assertEquals("ReduceTransform_7", lot.getOperators(7).getTransform().getId()); Assert.assertEquals("SEL_8", lot.getOperators(8).getSelect().getId()); Assert.assertEquals("DataSink_9", lot.getOperators(9).getDataSink().getId()); Assert.assertEquals(apsara.odps.TypesProtos.Type.String, lot.getOperators(1).getTransform() .getSchema().getColumns(0).getType()); Assert.assertEquals(apsara.odps.TypesProtos.Type.Integer, lot.getOperators(1).getTransform() .getSchema().getColumns(1).getType()); }
LogicalOperatorTree genTree() { resourceItems = buildResourceList(); LogicalOperatorTree.Builder builder = LogicalOperatorTree.newBuilder(); inputTableInfos = InputUtils.getTables(job); inputVolumeInfos = InputUtils.getVolumes(job); Map<TableInfoKey, List<LinkedHashMap<String, String>>> inputTables = mergeInputTableInfos(inputTableInfos); outputTableInfos = OutputUtils.getTables(job); outputVolumeInfos = OutputUtils.getVolumes(job); isNoOutput = outputTableInfos == null; isMultiInsert = !isNoOutput && outputTableInfos.length > 1; boolean isStreamingOutput = job.getNumReduceTasks() > 0 ? isStreamingReduce : isStreamingMap; List<OdpsType> outputColumnTypes = new ArrayList<OdpsType>(); if (isMultiInsert) { for (TableInfo ti : outputTableInfos) { List<OdpsType> tbColumnTypes = new ArrayList<OdpsType>(); for (Column col : job.getOutputSchema(ti.getLabel())) { tbColumnTypes.add(col.getType()); } int idx = Collections.indexOfSubList(outputColumnTypes, tbColumnTypes); if (idx >= 0) { outputIndexes.put(ti.getLabel(), idx); continue; } idx = outputColumns.size(); outputIndexes.put(ti.getLabel(), idx); for (Column col : job.getOutputSchema(ti.getLabel())) { String colName = "multiins" + idx + "_" + col.getName(); if (isStreamingOutput) { outputColumns.add(new Column(colName, OdpsType.STRING)); outputColumnTypes.add(OdpsType.STRING); } else { outputColumns.add(TypeUtils.createColumnWithNewName(colName, col)); outputColumnTypes.add(col.getType()); } } } outputColumns.add(new Column(MULTI_INSERT_SELECTOR, OdpsType.STRING)); } else if (isNoOutput) { outputColumns.add(new Column(NO_OUTPUT_DUMMY_COLUMN, OdpsType.STRING)); } else { for (Column col : job.getOutputSchema(outputTableInfos[0].getLabel())) { if (isStreamingOutput) { outputColumns.add(new Column(col.getName(), OdpsType.STRING)); } else { outputColumns.add(TypeUtils.cloneColumn(col)); } } } List<Column> mapOutColumns; List<Column> firstReduceInColumns = null; int innerOutputIndex = 0; if (hasReducer) { mapOutColumns = new ArrayList<Column>(); firstReduceInColumns = new ArrayList<Column>(); if (hasPartitioner) { mapOutColumns.add(new Column(PARTITION_ID, OdpsType.BIGINT)); } Column[] keys = this.pipeMode ? pipeline.getFirstNode().getOutputKeySchema() : job.getMapOutputKeySchema() ; for (Column col : keys) { Column keyCol = TypeUtils.createColumnWithNewName(MAP_OUT_KEY_PREFIX + col.getName(), col); mapOutColumns.add(keyCol); firstReduceInColumns.add(keyCol); } Column[] values = this.pipeMode ? pipeline.getFirstNode().getOutputValueSchema() : job.getMapOutputValueSchema(); for (Column col : values) { Column valCol = TypeUtils.createColumnWithNewName(MAP_OUT_VAL_PREFIX + col.getName(), col); mapOutColumns.add(valCol); firstReduceInColumns.add(valCol); } } else { mapOutColumns = outputColumns; } String mapperId = genMapBlock(builder, inputTables, mapOutColumns, innerOutputIndex, hasReducer && isInnerOutput && (inputTables.size() <= 1)); if (hasReducer) { genReduceBlock(builder, firstReduceInColumns, mapperId); } else { handleOutput(builder, false, outputColumns, mapperId, isTableOverwrite, innerOutputIndex); } return builder.build(); }
LOTGenerator { LogicalOperatorTree genTree() { resourceItems = buildResourceList(); LogicalOperatorTree.Builder builder = LogicalOperatorTree.newBuilder(); inputTableInfos = InputUtils.getTables(job); inputVolumeInfos = InputUtils.getVolumes(job); Map<TableInfoKey, List<LinkedHashMap<String, String>>> inputTables = mergeInputTableInfos(inputTableInfos); outputTableInfos = OutputUtils.getTables(job); outputVolumeInfos = OutputUtils.getVolumes(job); isNoOutput = outputTableInfos == null; isMultiInsert = !isNoOutput && outputTableInfos.length > 1; boolean isStreamingOutput = job.getNumReduceTasks() > 0 ? isStreamingReduce : isStreamingMap; List<OdpsType> outputColumnTypes = new ArrayList<OdpsType>(); if (isMultiInsert) { for (TableInfo ti : outputTableInfos) { List<OdpsType> tbColumnTypes = new ArrayList<OdpsType>(); for (Column col : job.getOutputSchema(ti.getLabel())) { tbColumnTypes.add(col.getType()); } int idx = Collections.indexOfSubList(outputColumnTypes, tbColumnTypes); if (idx >= 0) { outputIndexes.put(ti.getLabel(), idx); continue; } idx = outputColumns.size(); outputIndexes.put(ti.getLabel(), idx); for (Column col : job.getOutputSchema(ti.getLabel())) { String colName = "multiins" + idx + "_" + col.getName(); if (isStreamingOutput) { outputColumns.add(new Column(colName, OdpsType.STRING)); outputColumnTypes.add(OdpsType.STRING); } else { outputColumns.add(TypeUtils.createColumnWithNewName(colName, col)); outputColumnTypes.add(col.getType()); } } } outputColumns.add(new Column(MULTI_INSERT_SELECTOR, OdpsType.STRING)); } else if (isNoOutput) { outputColumns.add(new Column(NO_OUTPUT_DUMMY_COLUMN, OdpsType.STRING)); } else { for (Column col : job.getOutputSchema(outputTableInfos[0].getLabel())) { if (isStreamingOutput) { outputColumns.add(new Column(col.getName(), OdpsType.STRING)); } else { outputColumns.add(TypeUtils.cloneColumn(col)); } } } List<Column> mapOutColumns; List<Column> firstReduceInColumns = null; int innerOutputIndex = 0; if (hasReducer) { mapOutColumns = new ArrayList<Column>(); firstReduceInColumns = new ArrayList<Column>(); if (hasPartitioner) { mapOutColumns.add(new Column(PARTITION_ID, OdpsType.BIGINT)); } Column[] keys = this.pipeMode ? pipeline.getFirstNode().getOutputKeySchema() : job.getMapOutputKeySchema() ; for (Column col : keys) { Column keyCol = TypeUtils.createColumnWithNewName(MAP_OUT_KEY_PREFIX + col.getName(), col); mapOutColumns.add(keyCol); firstReduceInColumns.add(keyCol); } Column[] values = this.pipeMode ? pipeline.getFirstNode().getOutputValueSchema() : job.getMapOutputValueSchema(); for (Column col : values) { Column valCol = TypeUtils.createColumnWithNewName(MAP_OUT_VAL_PREFIX + col.getName(), col); mapOutColumns.add(valCol); firstReduceInColumns.add(valCol); } } else { mapOutColumns = outputColumns; } String mapperId = genMapBlock(builder, inputTables, mapOutColumns, innerOutputIndex, hasReducer && isInnerOutput && (inputTables.size() <= 1)); if (hasReducer) { genReduceBlock(builder, firstReduceInColumns, mapperId); } else { handleOutput(builder, false, outputColumns, mapperId, isTableOverwrite, innerOutputIndex); } return builder.build(); } }
LOTGenerator { LogicalOperatorTree genTree() { resourceItems = buildResourceList(); LogicalOperatorTree.Builder builder = LogicalOperatorTree.newBuilder(); inputTableInfos = InputUtils.getTables(job); inputVolumeInfos = InputUtils.getVolumes(job); Map<TableInfoKey, List<LinkedHashMap<String, String>>> inputTables = mergeInputTableInfos(inputTableInfos); outputTableInfos = OutputUtils.getTables(job); outputVolumeInfos = OutputUtils.getVolumes(job); isNoOutput = outputTableInfos == null; isMultiInsert = !isNoOutput && outputTableInfos.length > 1; boolean isStreamingOutput = job.getNumReduceTasks() > 0 ? isStreamingReduce : isStreamingMap; List<OdpsType> outputColumnTypes = new ArrayList<OdpsType>(); if (isMultiInsert) { for (TableInfo ti : outputTableInfos) { List<OdpsType> tbColumnTypes = new ArrayList<OdpsType>(); for (Column col : job.getOutputSchema(ti.getLabel())) { tbColumnTypes.add(col.getType()); } int idx = Collections.indexOfSubList(outputColumnTypes, tbColumnTypes); if (idx >= 0) { outputIndexes.put(ti.getLabel(), idx); continue; } idx = outputColumns.size(); outputIndexes.put(ti.getLabel(), idx); for (Column col : job.getOutputSchema(ti.getLabel())) { String colName = "multiins" + idx + "_" + col.getName(); if (isStreamingOutput) { outputColumns.add(new Column(colName, OdpsType.STRING)); outputColumnTypes.add(OdpsType.STRING); } else { outputColumns.add(TypeUtils.createColumnWithNewName(colName, col)); outputColumnTypes.add(col.getType()); } } } outputColumns.add(new Column(MULTI_INSERT_SELECTOR, OdpsType.STRING)); } else if (isNoOutput) { outputColumns.add(new Column(NO_OUTPUT_DUMMY_COLUMN, OdpsType.STRING)); } else { for (Column col : job.getOutputSchema(outputTableInfos[0].getLabel())) { if (isStreamingOutput) { outputColumns.add(new Column(col.getName(), OdpsType.STRING)); } else { outputColumns.add(TypeUtils.cloneColumn(col)); } } } List<Column> mapOutColumns; List<Column> firstReduceInColumns = null; int innerOutputIndex = 0; if (hasReducer) { mapOutColumns = new ArrayList<Column>(); firstReduceInColumns = new ArrayList<Column>(); if (hasPartitioner) { mapOutColumns.add(new Column(PARTITION_ID, OdpsType.BIGINT)); } Column[] keys = this.pipeMode ? pipeline.getFirstNode().getOutputKeySchema() : job.getMapOutputKeySchema() ; for (Column col : keys) { Column keyCol = TypeUtils.createColumnWithNewName(MAP_OUT_KEY_PREFIX + col.getName(), col); mapOutColumns.add(keyCol); firstReduceInColumns.add(keyCol); } Column[] values = this.pipeMode ? pipeline.getFirstNode().getOutputValueSchema() : job.getMapOutputValueSchema(); for (Column col : values) { Column valCol = TypeUtils.createColumnWithNewName(MAP_OUT_VAL_PREFIX + col.getName(), col); mapOutColumns.add(valCol); firstReduceInColumns.add(valCol); } } else { mapOutColumns = outputColumns; } String mapperId = genMapBlock(builder, inputTables, mapOutColumns, innerOutputIndex, hasReducer && isInnerOutput && (inputTables.size() <= 1)); if (hasReducer) { genReduceBlock(builder, firstReduceInColumns, mapperId); } else { handleOutput(builder, false, outputColumns, mapperId, isTableOverwrite, innerOutputIndex); } return builder.build(); } LOTGenerator(String project, JobConf job, Pipeline pipeline); }
LOTGenerator { LogicalOperatorTree genTree() { resourceItems = buildResourceList(); LogicalOperatorTree.Builder builder = LogicalOperatorTree.newBuilder(); inputTableInfos = InputUtils.getTables(job); inputVolumeInfos = InputUtils.getVolumes(job); Map<TableInfoKey, List<LinkedHashMap<String, String>>> inputTables = mergeInputTableInfos(inputTableInfos); outputTableInfos = OutputUtils.getTables(job); outputVolumeInfos = OutputUtils.getVolumes(job); isNoOutput = outputTableInfos == null; isMultiInsert = !isNoOutput && outputTableInfos.length > 1; boolean isStreamingOutput = job.getNumReduceTasks() > 0 ? isStreamingReduce : isStreamingMap; List<OdpsType> outputColumnTypes = new ArrayList<OdpsType>(); if (isMultiInsert) { for (TableInfo ti : outputTableInfos) { List<OdpsType> tbColumnTypes = new ArrayList<OdpsType>(); for (Column col : job.getOutputSchema(ti.getLabel())) { tbColumnTypes.add(col.getType()); } int idx = Collections.indexOfSubList(outputColumnTypes, tbColumnTypes); if (idx >= 0) { outputIndexes.put(ti.getLabel(), idx); continue; } idx = outputColumns.size(); outputIndexes.put(ti.getLabel(), idx); for (Column col : job.getOutputSchema(ti.getLabel())) { String colName = "multiins" + idx + "_" + col.getName(); if (isStreamingOutput) { outputColumns.add(new Column(colName, OdpsType.STRING)); outputColumnTypes.add(OdpsType.STRING); } else { outputColumns.add(TypeUtils.createColumnWithNewName(colName, col)); outputColumnTypes.add(col.getType()); } } } outputColumns.add(new Column(MULTI_INSERT_SELECTOR, OdpsType.STRING)); } else if (isNoOutput) { outputColumns.add(new Column(NO_OUTPUT_DUMMY_COLUMN, OdpsType.STRING)); } else { for (Column col : job.getOutputSchema(outputTableInfos[0].getLabel())) { if (isStreamingOutput) { outputColumns.add(new Column(col.getName(), OdpsType.STRING)); } else { outputColumns.add(TypeUtils.cloneColumn(col)); } } } List<Column> mapOutColumns; List<Column> firstReduceInColumns = null; int innerOutputIndex = 0; if (hasReducer) { mapOutColumns = new ArrayList<Column>(); firstReduceInColumns = new ArrayList<Column>(); if (hasPartitioner) { mapOutColumns.add(new Column(PARTITION_ID, OdpsType.BIGINT)); } Column[] keys = this.pipeMode ? pipeline.getFirstNode().getOutputKeySchema() : job.getMapOutputKeySchema() ; for (Column col : keys) { Column keyCol = TypeUtils.createColumnWithNewName(MAP_OUT_KEY_PREFIX + col.getName(), col); mapOutColumns.add(keyCol); firstReduceInColumns.add(keyCol); } Column[] values = this.pipeMode ? pipeline.getFirstNode().getOutputValueSchema() : job.getMapOutputValueSchema(); for (Column col : values) { Column valCol = TypeUtils.createColumnWithNewName(MAP_OUT_VAL_PREFIX + col.getName(), col); mapOutColumns.add(valCol); firstReduceInColumns.add(valCol); } } else { mapOutColumns = outputColumns; } String mapperId = genMapBlock(builder, inputTables, mapOutColumns, innerOutputIndex, hasReducer && isInnerOutput && (inputTables.size() <= 1)); if (hasReducer) { genReduceBlock(builder, firstReduceInColumns, mapperId); } else { handleOutput(builder, false, outputColumns, mapperId, isTableOverwrite, innerOutputIndex); } return builder.build(); } LOTGenerator(String project, JobConf job, Pipeline pipeline); byte[] generate(); }
LOTGenerator { LogicalOperatorTree genTree() { resourceItems = buildResourceList(); LogicalOperatorTree.Builder builder = LogicalOperatorTree.newBuilder(); inputTableInfos = InputUtils.getTables(job); inputVolumeInfos = InputUtils.getVolumes(job); Map<TableInfoKey, List<LinkedHashMap<String, String>>> inputTables = mergeInputTableInfos(inputTableInfos); outputTableInfos = OutputUtils.getTables(job); outputVolumeInfos = OutputUtils.getVolumes(job); isNoOutput = outputTableInfos == null; isMultiInsert = !isNoOutput && outputTableInfos.length > 1; boolean isStreamingOutput = job.getNumReduceTasks() > 0 ? isStreamingReduce : isStreamingMap; List<OdpsType> outputColumnTypes = new ArrayList<OdpsType>(); if (isMultiInsert) { for (TableInfo ti : outputTableInfos) { List<OdpsType> tbColumnTypes = new ArrayList<OdpsType>(); for (Column col : job.getOutputSchema(ti.getLabel())) { tbColumnTypes.add(col.getType()); } int idx = Collections.indexOfSubList(outputColumnTypes, tbColumnTypes); if (idx >= 0) { outputIndexes.put(ti.getLabel(), idx); continue; } idx = outputColumns.size(); outputIndexes.put(ti.getLabel(), idx); for (Column col : job.getOutputSchema(ti.getLabel())) { String colName = "multiins" + idx + "_" + col.getName(); if (isStreamingOutput) { outputColumns.add(new Column(colName, OdpsType.STRING)); outputColumnTypes.add(OdpsType.STRING); } else { outputColumns.add(TypeUtils.createColumnWithNewName(colName, col)); outputColumnTypes.add(col.getType()); } } } outputColumns.add(new Column(MULTI_INSERT_SELECTOR, OdpsType.STRING)); } else if (isNoOutput) { outputColumns.add(new Column(NO_OUTPUT_DUMMY_COLUMN, OdpsType.STRING)); } else { for (Column col : job.getOutputSchema(outputTableInfos[0].getLabel())) { if (isStreamingOutput) { outputColumns.add(new Column(col.getName(), OdpsType.STRING)); } else { outputColumns.add(TypeUtils.cloneColumn(col)); } } } List<Column> mapOutColumns; List<Column> firstReduceInColumns = null; int innerOutputIndex = 0; if (hasReducer) { mapOutColumns = new ArrayList<Column>(); firstReduceInColumns = new ArrayList<Column>(); if (hasPartitioner) { mapOutColumns.add(new Column(PARTITION_ID, OdpsType.BIGINT)); } Column[] keys = this.pipeMode ? pipeline.getFirstNode().getOutputKeySchema() : job.getMapOutputKeySchema() ; for (Column col : keys) { Column keyCol = TypeUtils.createColumnWithNewName(MAP_OUT_KEY_PREFIX + col.getName(), col); mapOutColumns.add(keyCol); firstReduceInColumns.add(keyCol); } Column[] values = this.pipeMode ? pipeline.getFirstNode().getOutputValueSchema() : job.getMapOutputValueSchema(); for (Column col : values) { Column valCol = TypeUtils.createColumnWithNewName(MAP_OUT_VAL_PREFIX + col.getName(), col); mapOutColumns.add(valCol); firstReduceInColumns.add(valCol); } } else { mapOutColumns = outputColumns; } String mapperId = genMapBlock(builder, inputTables, mapOutColumns, innerOutputIndex, hasReducer && isInnerOutput && (inputTables.size() <= 1)); if (hasReducer) { genReduceBlock(builder, firstReduceInColumns, mapperId); } else { handleOutput(builder, false, outputColumns, mapperId, isTableOverwrite, innerOutputIndex); } return builder.build(); } LOTGenerator(String project, JobConf job, Pipeline pipeline); byte[] generate();  }