@Test public void testSchemaValidation() { Transform<StructuredRecord, StructuredRecord> transform = new Decompressor(new Decompressor.Config("a:ZIP", OUTPUT.toString())); MockPipelineConfigurer mockPipelineConfigurer = new MockPipelineConfigurer(INPUT); transform.configurePipeline(mockPipelineConfigurer); Assert.assertEquals(OUTPUT, mockPipelineConfigurer.getOutputSchema()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.decompressor, collector); Schema outputSchema = config.getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); for (Map.Entry<String, DecompressorType> entry : deCompMap.entrySet()) { String fieldName = entry.getKey(); if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Decompress field '%s' must be present in output schema.", fieldName), "Add decompress field in output schema or remove it from decompress.") .withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); continue; } Schema fieldSchema = outputSchema.getField(fieldName).getSchema(); fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema; if (outSchemaMap.get(fieldName) != Schema.Type.BYTES && outSchemaMap.get(fieldName) != Schema.Type.STRING) { collector.addFailure(String.format("Field '%s' is of unexpected type '%s'.", fieldName, fieldSchema.getDisplayName()), "It must be of type 'bytes' or 'string'") .withOutputSchemaField(fieldName).withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); }
Decompressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.decompressor, collector); Schema outputSchema = config.getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); for (Map.Entry<String, DecompressorType> entry : deCompMap.entrySet()) { String fieldName = entry.getKey(); if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Decompress field '%s' must be present in output schema.", fieldName), "Add decompress field in output schema or remove it from decompress.") .withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); continue; } Schema fieldSchema = outputSchema.getField(fieldName).getSchema(); fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema; if (outSchemaMap.get(fieldName) != Schema.Type.BYTES && outSchemaMap.get(fieldName) != Schema.Type.STRING) { collector.addFailure(String.format("Field '%s' is of unexpected type '%s'.", fieldName, fieldSchema.getDisplayName()), "It must be of type 'bytes' or 'string'") .withOutputSchemaField(fieldName).withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } }
Decompressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.decompressor, collector); Schema outputSchema = config.getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); for (Map.Entry<String, DecompressorType> entry : deCompMap.entrySet()) { String fieldName = entry.getKey(); if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Decompress field '%s' must be present in output schema.", fieldName), "Add decompress field in output schema or remove it from decompress.") .withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); continue; } Schema fieldSchema = outputSchema.getField(fieldName).getSchema(); fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema; if (outSchemaMap.get(fieldName) != Schema.Type.BYTES && outSchemaMap.get(fieldName) != Schema.Type.STRING) { collector.addFailure(String.format("Field '%s' is of unexpected type '%s'.", fieldName, fieldSchema.getDisplayName()), "It must be of type 'bytes' or 'string'") .withOutputSchemaField(fieldName).withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } Decompressor(Config config); }
Decompressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.decompressor, collector); Schema outputSchema = config.getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); for (Map.Entry<String, DecompressorType> entry : deCompMap.entrySet()) { String fieldName = entry.getKey(); if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Decompress field '%s' must be present in output schema.", fieldName), "Add decompress field in output schema or remove it from decompress.") .withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); continue; } Schema fieldSchema = outputSchema.getField(fieldName).getSchema(); fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema; if (outSchemaMap.get(fieldName) != Schema.Type.BYTES && outSchemaMap.get(fieldName) != Schema.Type.STRING) { collector.addFailure(String.format("Field '%s' is of unexpected type '%s'.", fieldName, fieldSchema.getDisplayName()), "It must be of type 'bytes' or 'string'") .withOutputSchemaField(fieldName).withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } Decompressor(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); }
Decompressor extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) throws IllegalArgumentException { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); parseConfiguration(config.decompressor, collector); Schema outputSchema = config.getSchema(collector); List<Field> outFields = outputSchema.getFields(); for (Field field : outFields) { outSchemaMap.put(field.getName(), field.getSchema().getType()); } validateInputSchema(pipelineConfigurer.getStageConfigurer().getInputSchema(), collector); for (Map.Entry<String, DecompressorType> entry : deCompMap.entrySet()) { String fieldName = entry.getKey(); if (!outSchemaMap.containsKey(fieldName)) { collector.addFailure(String.format("Decompress field '%s' must be present in output schema.", fieldName), "Add decompress field in output schema or remove it from decompress.") .withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); continue; } Schema fieldSchema = outputSchema.getField(fieldName).getSchema(); fieldSchema = fieldSchema.isNullable() ? fieldSchema.getNonNullable() : fieldSchema; if (outSchemaMap.get(fieldName) != Schema.Type.BYTES && outSchemaMap.get(fieldName) != Schema.Type.STRING) { collector.addFailure(String.format("Field '%s' is of unexpected type '%s'.", fieldName, fieldSchema.getDisplayName()), "It must be of type 'bytes' or 'string'") .withOutputSchemaField(fieldName).withConfigElement(Config.NAME_DECOMPRESSOR, fieldName + Config.SEPARATOR + entry.getValue()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } Decompressor(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);  }