CSVParser extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer configurer) throws IllegalArgumentException { super.configurePipeline(configurer); StageConfigurer stageConfigurer = configurer.getStageConfigurer(); FailureCollector collector = stageConfigurer.getFailureCollector(); config.validate(collector); Schema inputSchema = stageConfigurer.getInputSchema(); validateInputSchema(inputSchema, collector); Schema schema = parseAndValidateOutputSchema(inputSchema, collector); collector.getOrThrowException(); stageConfigurer.setOutputSchema(schema); } CSVParser(Config config); @Override void configurePipeline(PipelineConfigurer configurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord in, Emitter<StructuredRecord> emitter); static final CSVFormat PDL; }
@Test public void testPassThroughTypeMisMatch() { CSVParser.Config config = new CSVParser.Config("DEFAULT", null, "body", OUTPUT4.toString()); Transform<StructuredRecord, StructuredRecord> transform = new CSVParser(config); MockPipelineConfigurer configurer = new MockPipelineConfigurer(INPUT2); try { transform.configurePipeline(configurer); } catch (ValidationException e) { Assert.assertEquals(1, configurer.getStageConfigurer().getFailureCollector().getValidationFailures().size()); } }
@Test public void testEmptyCustomDelimiter() { CSVParser config = new CSVParser(new CSVParser.Config("Custom", null, "body", OUTPUT4.toString())); MockPipelineConfigurer configurer = new MockPipelineConfigurer(INPUT2); try { config.configurePipeline(configurer); Assert.fail(); } catch (ValidationException e) { Assert.assertEquals(2, configurer.getStageConfigurer().getFailureCollector().getValidationFailures().size()); } }