@Test public void testSchemaValidationWithValidInputSchema() { Transform<StructuredRecord, StructuredRecord> transform = new Compressor(new Compressor.Config("a:NONE", OUTPUT.toString())); Schema validInput = Schema.recordOf("input", Schema.Field.of("a", Schema.of(Schema.Type.INT)), Schema.Field.of("b", Schema.of(Schema.Type.STRING))); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(validInput); transform.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(OUTPUT, mockPipelineConfigurer.getOutputSchema()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.compressor, collector); Schema outputSchema = getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } for (String fieldName : compMap.keySet()) { if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Compress field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the compress field is of type bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); }
Compressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.compressor, collector); Schema outputSchema = getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } for (String fieldName : compMap.keySet()) { if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Compress field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the compress field is of type bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); } }
Compressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.compressor, collector); Schema outputSchema = getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } for (String fieldName : compMap.keySet()) { if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Compress field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the compress field is of type bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); } Compressor(Config config); }
Compressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.compressor, collector); Schema outputSchema = getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } for (String fieldName : compMap.keySet()) { if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Compress field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the compress field is of type bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); } Compressor(Config config); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord in, Emitter<StructuredRecord> emitter); }
Compressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.compressor, collector); Schema outputSchema = getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } for (String fieldName : compMap.keySet()) { if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Field '%s' must be in output schema.", fieldName), null) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); continue; } Field field = outputSchema.getField(fieldName); Schema nonNullableSchema = field.getSchema().isNullable() ? field.getSchema().getNonNullable() : field.getSchema(); if (nonNullableSchema.getType() != Schema.Type.BYTES || nonNullableSchema.getLogicalType() != null) { collector.addFailure(String.format("Compress field '%s' is of invalid type '%s'.", field.getName(), nonNullableSchema.getDisplayName()), "Ensure the compress field is of type bytes.") .withOutputSchemaField(field.getName()) .withConfigElement(Config.NAME_COMPRESSOR, fieldName + Config.SEPARATOR + compMap.get(fieldName)); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); } Compressor(Config config); @Override void prepareRun(StageSubmitterContext context); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord in, Emitter<StructuredRecord> emitter);  }