@Test public void testNormalize() throws Exception { String inputTable = "inputNormalizeTable"; Map<String, String> sourceproperties = new ImmutableMap.Builder<String, String>() .put("fieldMapping", validFieldMapping) .put("fieldNormalizing", validFieldNormalizing) .put("outputSchema", OUTPUT_SCHEMA.toString()) .build(); String outputTable = "outputNormalizeTable"; ApplicationManager applicationManager = deployApplication(sourceproperties, inputTable, outputTable, "normalizeTest"); DataSetManager<Table> inputManager = getDataset(inputTable); List<StructuredRecord> input = ImmutableList.of( StructuredRecord.builder(INPUT_SCHEMA).set(ITEM_ID, ITEM_ID_ROW1).set(CUSTOMER_ID, CUSTOMER_ID_FIRST) .set(ITEM_COST, ITEM_COST_ROW1).set(PURCHASE_DATE, PURCHASE_DATE_ROW1).build(), StructuredRecord.builder(INPUT_SCHEMA).set(ITEM_ID, ITEM_ID_ROW2).set(CUSTOMER_ID, CUSTOMER_ID_FIRST) .set(ITEM_COST, ITEM_COST_ROW2).set(PURCHASE_DATE, PURCHASE_DATE_ROW2).build(), StructuredRecord.builder(INPUT_SCHEMA).set(ITEM_ID, ITEM_ID_ROW3).set(CUSTOMER_ID, CUSTOMER_ID_SECOND) .set(ITEM_COST, ITEM_COST_ROW3).set(PURCHASE_DATE, PURCHASE_DATE_ROW3).build() ); MockSource.writeInput(inputManager, input); startWorkflow(applicationManager, ProgramRunStatus.COMPLETED); DataSetManager<Table> outputManager = getDataset(outputTable); List<StructuredRecord> outputRecords = MockSink.readOutput(outputManager); Assert.assertEquals(6, outputRecords.size()); Assert.assertEquals(outputRecords.get(0).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(0)))); Assert.assertEquals(outputRecords.get(1).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(1)))); Assert.assertEquals(outputRecords.get(2).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(2)))); Assert.assertEquals(outputRecords.get(3).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(3)))); Assert.assertEquals(outputRecords.get(4).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(4)))); Assert.assertEquals(outputRecords.get(5).get(ATTRIBUTE_VALUE), dataMap.get(getKeyFromRecord(outputRecords.get(5)))); }
public Normalize(NormalizeConfig config) { this.config = config; }
Normalize extends Transform<StructuredRecord, StructuredRecord> { public Normalize(NormalizeConfig config) { this.config = config; } }
Normalize extends Transform<StructuredRecord, StructuredRecord> { public Normalize(NormalizeConfig config) { this.config = config; } Normalize(NormalizeConfig config); }
Normalize extends Transform<StructuredRecord, StructuredRecord> { public Normalize(NormalizeConfig config) { this.config = config; } Normalize(NormalizeConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord structuredRecord, Emitter<StructuredRecord> emitter); }
Normalize extends Transform<StructuredRecord, StructuredRecord> { public Normalize(NormalizeConfig config) { this.config = config; } Normalize(NormalizeConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord structuredRecord, Emitter<StructuredRecord> emitter);  }