@Test public void testSchemaValidation() throws Exception { JSONFormatter.Config config = new JSONFormatter.Config(OUTPUT1.toString()); Transform<StructuredRecord, StructuredRecord> transform = new JSONFormatter(config); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(INPUT1); transform.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(OUTPUT1, mockPipelineConfigurer.getOutputSchema()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema out = Schema.parseJson(config.schema); List<Schema.Field> fields = out.getFields(); if (fields.size() > 1) { collector.addFailure("Output schema may only have one field.", null); } Schema.Field field = fields.get(0); Schema.Type type = field.getSchema().getType(); if (fields.get(0).getSchema().getType() != Schema.Type.STRING && fields.get(0).getSchema().getType() != Schema.Type.BYTES) { collector.addFailure(String.format("Invalid output field type '%s' for field '%s'.", field.getName(), type.toString().toLowerCase()), "Output field type must be of type string or bytes.") .withOutputSchemaField(field.getName()); } pipelineConfigurer.getStageConfigurer().setOutputSchema(out); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(Config.SCHEMA); } }
JSONFormatter extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema out = Schema.parseJson(config.schema); List<Schema.Field> fields = out.getFields(); if (fields.size() > 1) { collector.addFailure("Output schema may only have one field.", null); } Schema.Field field = fields.get(0); Schema.Type type = field.getSchema().getType(); if (fields.get(0).getSchema().getType() != Schema.Type.STRING && fields.get(0).getSchema().getType() != Schema.Type.BYTES) { collector.addFailure(String.format("Invalid output field type '%s' for field '%s'.", field.getName(), type.toString().toLowerCase()), "Output field type must be of type string or bytes.") .withOutputSchemaField(field.getName()); } pipelineConfigurer.getStageConfigurer().setOutputSchema(out); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(Config.SCHEMA); } } }
JSONFormatter extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema out = Schema.parseJson(config.schema); List<Schema.Field> fields = out.getFields(); if (fields.size() > 1) { collector.addFailure("Output schema may only have one field.", null); } Schema.Field field = fields.get(0); Schema.Type type = field.getSchema().getType(); if (fields.get(0).getSchema().getType() != Schema.Type.STRING && fields.get(0).getSchema().getType() != Schema.Type.BYTES) { collector.addFailure(String.format("Invalid output field type '%s' for field '%s'.", field.getName(), type.toString().toLowerCase()), "Output field type must be of type string or bytes.") .withOutputSchemaField(field.getName()); } pipelineConfigurer.getStageConfigurer().setOutputSchema(out); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(Config.SCHEMA); } } JSONFormatter(Config config); }
JSONFormatter extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema out = Schema.parseJson(config.schema); List<Schema.Field> fields = out.getFields(); if (fields.size() > 1) { collector.addFailure("Output schema may only have one field.", null); } Schema.Field field = fields.get(0); Schema.Type type = field.getSchema().getType(); if (fields.get(0).getSchema().getType() != Schema.Type.STRING && fields.get(0).getSchema().getType() != Schema.Type.BYTES) { collector.addFailure(String.format("Invalid output field type '%s' for field '%s'.", field.getName(), type.toString().toLowerCase()), "Output field type must be of type string or bytes.") .withOutputSchemaField(field.getName()); } pipelineConfigurer.getStageConfigurer().setOutputSchema(out); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(Config.SCHEMA); } } JSONFormatter(Config config); @Override void initialize(TransformContext context); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter); }
JSONFormatter extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); try { Schema out = Schema.parseJson(config.schema); List<Schema.Field> fields = out.getFields(); if (fields.size() > 1) { collector.addFailure("Output schema may only have one field.", null); } Schema.Field field = fields.get(0); Schema.Type type = field.getSchema().getType(); if (fields.get(0).getSchema().getType() != Schema.Type.STRING && fields.get(0).getSchema().getType() != Schema.Type.BYTES) { collector.addFailure(String.format("Invalid output field type '%s' for field '%s'.", field.getName(), type.toString().toLowerCase()), "Output field type must be of type string or bytes.") .withOutputSchemaField(field.getName()); } pipelineConfigurer.getStageConfigurer().setOutputSchema(out); } catch (IOException e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(Config.SCHEMA); } } JSONFormatter(Config config); @Override void initialize(TransformContext context); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void transform(StructuredRecord input, Emitter<StructuredRecord> emitter);  }