@Test public void testMappingValidation() throws Exception { Schema inputSchema = Schema.recordOf("sourceRecord", Schema.Field.of(ID, Schema.of(Schema.Type.STRING)), Schema.Field.of(NAME, Schema.of(Schema.Type.STRING)), Schema.Field.of(SALARY, Schema.of(Schema.Type.STRING)), Schema.Field.of(DESIGNATIONID, Schema.of(Schema.Type.STRING))); ValueMapper.Config config = new ValueMapper.Config("designationid:designation_lookup_table", "designationid:DEFAULTID"); MockPipelineConfigurer configurer = new MockPipelineConfigurer(inputSchema); try { new ValueMapper(config).configurePipeline(configurer); Assert.fail(); } catch (ValidationException e) { Assert.assertEquals(1, e.getFailures().size()); Assert.assertEquals(1, e.getFailures().get(0).getCauses().size()); Cause expectedCause = new Cause(); expectedCause.addAttribute(STAGE, MOCK_STAGE); expectedCause.addAttribute(CauseAttributes.STAGE_CONFIG, ValueMapper.Config.MAPPING); expectedCause.addAttribute(CauseAttributes.CONFIG_ELEMENT, "designationid:designation_lookup_table"); Assert.assertEquals(expectedCause, e.getFailures().get(0).getCauses().get(0)); } }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(this.config, collector); super.configurePipeline(pipelineConfigurer); Schema outputSchema = null; Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Map.Entry<String, ValueMapping> mappingPair: mappingValues.entrySet()) { if (inputSchema.getField(mappingPair.getKey()) == null) { collector.addFailure( String.format("Map key '%s' must be present in the input schema.", mappingPair.getKey()), null) .withConfigElement( Config.MAPPING, String.format("%s:%s:%s", mappingPair.getKey(), mappingPair.getValue().getLookupTableName(), mappingPair.getValue().getTargetField())); } } for (Map.Entry<String, String> defaultPair: defaultsMapping.entrySet()) { if (!mappingValues.containsKey(defaultPair.getKey())) { collector.addFailure( String.format("Defaults key '%s' must be present as a source in mapping.", defaultPair.getKey()), null) .withConfigElement(Config.DEFAULTS, String.format("%s:%s", defaultPair.getKey(), defaultPair.getValue())); } } outputSchema = getOutputSchema(inputSchema, collector); } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); }
ValueMapper extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(this.config, collector); super.configurePipeline(pipelineConfigurer); Schema outputSchema = null; Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Map.Entry<String, ValueMapping> mappingPair: mappingValues.entrySet()) { if (inputSchema.getField(mappingPair.getKey()) == null) { collector.addFailure( String.format("Map key '%s' must be present in the input schema.", mappingPair.getKey()), null) .withConfigElement( Config.MAPPING, String.format("%s:%s:%s", mappingPair.getKey(), mappingPair.getValue().getLookupTableName(), mappingPair.getValue().getTargetField())); } } for (Map.Entry<String, String> defaultPair: defaultsMapping.entrySet()) { if (!mappingValues.containsKey(defaultPair.getKey())) { collector.addFailure( String.format("Defaults key '%s' must be present as a source in mapping.", defaultPair.getKey()), null) .withConfigElement(Config.DEFAULTS, String.format("%s:%s", defaultPair.getKey(), defaultPair.getValue())); } } outputSchema = getOutputSchema(inputSchema, collector); } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } }
ValueMapper extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(this.config, collector); super.configurePipeline(pipelineConfigurer); Schema outputSchema = null; Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Map.Entry<String, ValueMapping> mappingPair: mappingValues.entrySet()) { if (inputSchema.getField(mappingPair.getKey()) == null) { collector.addFailure( String.format("Map key '%s' must be present in the input schema.", mappingPair.getKey()), null) .withConfigElement( Config.MAPPING, String.format("%s:%s:%s", mappingPair.getKey(), mappingPair.getValue().getLookupTableName(), mappingPair.getValue().getTargetField())); } } for (Map.Entry<String, String> defaultPair: defaultsMapping.entrySet()) { if (!mappingValues.containsKey(defaultPair.getKey())) { collector.addFailure( String.format("Defaults key '%s' must be present as a source in mapping.", defaultPair.getKey()), null) .withConfigElement(Config.DEFAULTS, String.format("%s:%s", defaultPair.getKey(), defaultPair.getValue())); } } outputSchema = getOutputSchema(inputSchema, collector); } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } ValueMapper(Config config); }
ValueMapper extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(this.config, collector); super.configurePipeline(pipelineConfigurer); Schema outputSchema = null; Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Map.Entry<String, ValueMapping> mappingPair: mappingValues.entrySet()) { if (inputSchema.getField(mappingPair.getKey()) == null) { collector.addFailure( String.format("Map key '%s' must be present in the input schema.", mappingPair.getKey()), null) .withConfigElement( Config.MAPPING, String.format("%s:%s:%s", mappingPair.getKey(), mappingPair.getValue().getLookupTableName(), mappingPair.getValue().getTargetField())); } } for (Map.Entry<String, String> defaultPair: defaultsMapping.entrySet()) { if (!mappingValues.containsKey(defaultPair.getKey())) { collector.addFailure( String.format("Defaults key '%s' must be present as a source in mapping.", defaultPair.getKey()), null) .withConfigElement(Config.DEFAULTS, String.format("%s:%s", defaultPair.getKey(), defaultPair.getValue())); } } outputSchema = getOutputSchema(inputSchema, collector); } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } ValueMapper(Config config); @Override void prepareRun(StageSubmitterContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter); @Override void initialize(TransformContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); }
ValueMapper extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(this.config, collector); super.configurePipeline(pipelineConfigurer); Schema outputSchema = null; Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); if (inputSchema != null) { for (Map.Entry<String, ValueMapping> mappingPair: mappingValues.entrySet()) { if (inputSchema.getField(mappingPair.getKey()) == null) { collector.addFailure( String.format("Map key '%s' must be present in the input schema.", mappingPair.getKey()), null) .withConfigElement( Config.MAPPING, String.format("%s:%s:%s", mappingPair.getKey(), mappingPair.getValue().getLookupTableName(), mappingPair.getValue().getTargetField())); } } for (Map.Entry<String, String> defaultPair: defaultsMapping.entrySet()) { if (!mappingValues.containsKey(defaultPair.getKey())) { collector.addFailure( String.format("Defaults key '%s' must be present as a source in mapping.", defaultPair.getKey()), null) .withConfigElement(Config.DEFAULTS, String.format("%s:%s", defaultPair.getKey(), defaultPair.getValue())); } } outputSchema = getOutputSchema(inputSchema, collector); } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } ValueMapper(Config config); @Override void prepareRun(StageSubmitterContext context); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter); @Override void initialize(TransformContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer);  }