@Test public void testOutputSchema() throws Exception { Normalize.NormalizeConfig config = new Normalize.NormalizeConfig(validFieldMapping, validFieldNormalizing, OUTPUT_SCHEMA.toString()); MockPipelineConfigurer configurer = new MockPipelineConfigurer(INPUT_SCHEMA); new Normalize(config).configurePipeline(configurer); Assert.assertEquals(OUTPUT_SCHEMA, configurer.getOutputSchema()); }
@Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); config.validate(collector); collector.getOrThrowException(); try { outputSchema = Schema.parseJson(config.outputSchema); for (Schema.Field outputField : outputSchema.getFields()) { Schema fieldSchema = outputField.getSchema(); Schema.Type fieldType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!fieldType.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Invalid type '%s' for output schema field '%s'.", fieldType.toString(), outputField.getName()), "All output schema fields must be of type string.") .withOutputSchemaField(outputField.getName()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (Exception e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(NormalizeConfig.OUTPUT_SCHEMA); return; } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); List<String> fieldList = new ArrayList<>(); String[] fieldMappingArray = config.fieldMapping.split(","); for (String fieldMapping : fieldMappingArray) { String[] mappings = fieldMapping.split(":"); if (mappings.length != 2) { collector.addFailure("Field name and mapping must be provided.", null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); continue; } boolean failed = false; if (inputSchema != null && inputSchema.getField(mappings[0]) == null) { collector.addFailure(String.format("Field '%s' must be present in input schema.", mappings[0]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (outputSchema.getField(mappings[1]) == null) { collector.addFailure(String.format("Mapping '%s' must be present in output schema.", mappings[1]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (!failed) { fieldList.add(mappings[0]); } } String[] fieldNormalizingArray = config.fieldNormalizing.split(","); String validTypeField = null; String validValueField = null; for (String fieldNormalizing : fieldNormalizingArray) { String[] fields = fieldNormalizing.split(":"); if (fields.length != 3 || Strings.isNullOrEmpty(fields[0]) || Strings.isNullOrEmpty(fields[1]) || Strings.isNullOrEmpty(fields[2])) { collector.addFailure("Normalizing field, field name, and field value must all be provided.", null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); continue; } if (inputSchema != null && inputSchema.getField(fields[0]) == null) { collector.addFailure( String.format("Normalizing field '%s' must be present in input schema.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (fieldList.contains(fields[0])) { collector.addFailure( String.format("Field '%s' cannot be used for both mapping and normalizing.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validTypeField == null) { validTypeField = fields[1]; } else if (!validTypeField.equals(fields[1])) { collector.addFailure( String.format("Invalid normalizing field type output column name '%s'.", fields[1]), "All normalizing field type output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validValueField == null) { validValueField = fields[2]; } else if (!validValueField.equals(fields[2])) { collector.addFailure( String.format("Invalid normalizing field value output column name '%s'.", fields[2]), "All normalizing field value output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[1]) == null) { collector.addFailure( String.format("Normalizing field type output column name '%s' must be present in output schema.", fields[1]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[2]) == null) { collector.addFailure( String.format("Normalizing field value output column name '%s' must be present in output schema.", fields[2]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } } }
Normalize extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); config.validate(collector); collector.getOrThrowException(); try { outputSchema = Schema.parseJson(config.outputSchema); for (Schema.Field outputField : outputSchema.getFields()) { Schema fieldSchema = outputField.getSchema(); Schema.Type fieldType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!fieldType.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Invalid type '%s' for output schema field '%s'.", fieldType.toString(), outputField.getName()), "All output schema fields must be of type string.") .withOutputSchemaField(outputField.getName()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (Exception e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(NormalizeConfig.OUTPUT_SCHEMA); return; } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); List<String> fieldList = new ArrayList<>(); String[] fieldMappingArray = config.fieldMapping.split(","); for (String fieldMapping : fieldMappingArray) { String[] mappings = fieldMapping.split(":"); if (mappings.length != 2) { collector.addFailure("Field name and mapping must be provided.", null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); continue; } boolean failed = false; if (inputSchema != null && inputSchema.getField(mappings[0]) == null) { collector.addFailure(String.format("Field '%s' must be present in input schema.", mappings[0]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (outputSchema.getField(mappings[1]) == null) { collector.addFailure(String.format("Mapping '%s' must be present in output schema.", mappings[1]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (!failed) { fieldList.add(mappings[0]); } } String[] fieldNormalizingArray = config.fieldNormalizing.split(","); String validTypeField = null; String validValueField = null; for (String fieldNormalizing : fieldNormalizingArray) { String[] fields = fieldNormalizing.split(":"); if (fields.length != 3 || Strings.isNullOrEmpty(fields[0]) || Strings.isNullOrEmpty(fields[1]) || Strings.isNullOrEmpty(fields[2])) { collector.addFailure("Normalizing field, field name, and field value must all be provided.", null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); continue; } if (inputSchema != null && inputSchema.getField(fields[0]) == null) { collector.addFailure( String.format("Normalizing field '%s' must be present in input schema.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (fieldList.contains(fields[0])) { collector.addFailure( String.format("Field '%s' cannot be used for both mapping and normalizing.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validTypeField == null) { validTypeField = fields[1]; } else if (!validTypeField.equals(fields[1])) { collector.addFailure( String.format("Invalid normalizing field type output column name '%s'.", fields[1]), "All normalizing field type output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validValueField == null) { validValueField = fields[2]; } else if (!validValueField.equals(fields[2])) { collector.addFailure( String.format("Invalid normalizing field value output column name '%s'.", fields[2]), "All normalizing field value output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[1]) == null) { collector.addFailure( String.format("Normalizing field type output column name '%s' must be present in output schema.", fields[1]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[2]) == null) { collector.addFailure( String.format("Normalizing field value output column name '%s' must be present in output schema.", fields[2]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } } } }
Normalize extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); config.validate(collector); collector.getOrThrowException(); try { outputSchema = Schema.parseJson(config.outputSchema); for (Schema.Field outputField : outputSchema.getFields()) { Schema fieldSchema = outputField.getSchema(); Schema.Type fieldType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!fieldType.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Invalid type '%s' for output schema field '%s'.", fieldType.toString(), outputField.getName()), "All output schema fields must be of type string.") .withOutputSchemaField(outputField.getName()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (Exception e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(NormalizeConfig.OUTPUT_SCHEMA); return; } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); List<String> fieldList = new ArrayList<>(); String[] fieldMappingArray = config.fieldMapping.split(","); for (String fieldMapping : fieldMappingArray) { String[] mappings = fieldMapping.split(":"); if (mappings.length != 2) { collector.addFailure("Field name and mapping must be provided.", null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); continue; } boolean failed = false; if (inputSchema != null && inputSchema.getField(mappings[0]) == null) { collector.addFailure(String.format("Field '%s' must be present in input schema.", mappings[0]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (outputSchema.getField(mappings[1]) == null) { collector.addFailure(String.format("Mapping '%s' must be present in output schema.", mappings[1]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (!failed) { fieldList.add(mappings[0]); } } String[] fieldNormalizingArray = config.fieldNormalizing.split(","); String validTypeField = null; String validValueField = null; for (String fieldNormalizing : fieldNormalizingArray) { String[] fields = fieldNormalizing.split(":"); if (fields.length != 3 || Strings.isNullOrEmpty(fields[0]) || Strings.isNullOrEmpty(fields[1]) || Strings.isNullOrEmpty(fields[2])) { collector.addFailure("Normalizing field, field name, and field value must all be provided.", null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); continue; } if (inputSchema != null && inputSchema.getField(fields[0]) == null) { collector.addFailure( String.format("Normalizing field '%s' must be present in input schema.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (fieldList.contains(fields[0])) { collector.addFailure( String.format("Field '%s' cannot be used for both mapping and normalizing.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validTypeField == null) { validTypeField = fields[1]; } else if (!validTypeField.equals(fields[1])) { collector.addFailure( String.format("Invalid normalizing field type output column name '%s'.", fields[1]), "All normalizing field type output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validValueField == null) { validValueField = fields[2]; } else if (!validValueField.equals(fields[2])) { collector.addFailure( String.format("Invalid normalizing field value output column name '%s'.", fields[2]), "All normalizing field value output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[1]) == null) { collector.addFailure( String.format("Normalizing field type output column name '%s' must be present in output schema.", fields[1]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[2]) == null) { collector.addFailure( String.format("Normalizing field value output column name '%s' must be present in output schema.", fields[2]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } } } Normalize(NormalizeConfig config); }
Normalize extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); config.validate(collector); collector.getOrThrowException(); try { outputSchema = Schema.parseJson(config.outputSchema); for (Schema.Field outputField : outputSchema.getFields()) { Schema fieldSchema = outputField.getSchema(); Schema.Type fieldType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!fieldType.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Invalid type '%s' for output schema field '%s'.", fieldType.toString(), outputField.getName()), "All output schema fields must be of type string.") .withOutputSchemaField(outputField.getName()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (Exception e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(NormalizeConfig.OUTPUT_SCHEMA); return; } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); List<String> fieldList = new ArrayList<>(); String[] fieldMappingArray = config.fieldMapping.split(","); for (String fieldMapping : fieldMappingArray) { String[] mappings = fieldMapping.split(":"); if (mappings.length != 2) { collector.addFailure("Field name and mapping must be provided.", null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); continue; } boolean failed = false; if (inputSchema != null && inputSchema.getField(mappings[0]) == null) { collector.addFailure(String.format("Field '%s' must be present in input schema.", mappings[0]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (outputSchema.getField(mappings[1]) == null) { collector.addFailure(String.format("Mapping '%s' must be present in output schema.", mappings[1]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (!failed) { fieldList.add(mappings[0]); } } String[] fieldNormalizingArray = config.fieldNormalizing.split(","); String validTypeField = null; String validValueField = null; for (String fieldNormalizing : fieldNormalizingArray) { String[] fields = fieldNormalizing.split(":"); if (fields.length != 3 || Strings.isNullOrEmpty(fields[0]) || Strings.isNullOrEmpty(fields[1]) || Strings.isNullOrEmpty(fields[2])) { collector.addFailure("Normalizing field, field name, and field value must all be provided.", null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); continue; } if (inputSchema != null && inputSchema.getField(fields[0]) == null) { collector.addFailure( String.format("Normalizing field '%s' must be present in input schema.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (fieldList.contains(fields[0])) { collector.addFailure( String.format("Field '%s' cannot be used for both mapping and normalizing.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validTypeField == null) { validTypeField = fields[1]; } else if (!validTypeField.equals(fields[1])) { collector.addFailure( String.format("Invalid normalizing field type output column name '%s'.", fields[1]), "All normalizing field type output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validValueField == null) { validValueField = fields[2]; } else if (!validValueField.equals(fields[2])) { collector.addFailure( String.format("Invalid normalizing field value output column name '%s'.", fields[2]), "All normalizing field value output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[1]) == null) { collector.addFailure( String.format("Normalizing field type output column name '%s' must be present in output schema.", fields[1]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[2]) == null) { collector.addFailure( String.format("Normalizing field value output column name '%s' must be present in output schema.", fields[2]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } } } Normalize(NormalizeConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord structuredRecord, Emitter<StructuredRecord> emitter); }
Normalize extends Transform<StructuredRecord, StructuredRecord> { @Override public void configurePipeline(PipelineConfigurer pipelineConfigurer) { super.configurePipeline(pipelineConfigurer); FailureCollector collector = pipelineConfigurer.getStageConfigurer().getFailureCollector(); config.validate(collector); collector.getOrThrowException(); try { outputSchema = Schema.parseJson(config.outputSchema); for (Schema.Field outputField : outputSchema.getFields()) { Schema fieldSchema = outputField.getSchema(); Schema.Type fieldType = fieldSchema.isNullable() ? fieldSchema.getNonNullable().getType() : fieldSchema.getType(); if (!fieldType.equals(Schema.Type.STRING)) { collector.addFailure(String.format("Invalid type '%s' for output schema field '%s'.", fieldType.toString(), outputField.getName()), "All output schema fields must be of type string.") .withOutputSchemaField(outputField.getName()); } } pipelineConfigurer.getStageConfigurer().setOutputSchema(outputSchema); } catch (Exception e) { collector.addFailure("Invalid output schema.", "Output schema must be valid JSON.") .withConfigProperty(NormalizeConfig.OUTPUT_SCHEMA); return; } Schema inputSchema = pipelineConfigurer.getStageConfigurer().getInputSchema(); List<String> fieldList = new ArrayList<>(); String[] fieldMappingArray = config.fieldMapping.split(","); for (String fieldMapping : fieldMappingArray) { String[] mappings = fieldMapping.split(":"); if (mappings.length != 2) { collector.addFailure("Field name and mapping must be provided.", null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); continue; } boolean failed = false; if (inputSchema != null && inputSchema.getField(mappings[0]) == null) { collector.addFailure(String.format("Field '%s' must be present in input schema.", mappings[0]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (outputSchema.getField(mappings[1]) == null) { collector.addFailure(String.format("Mapping '%s' must be present in output schema.", mappings[1]), null) .withConfigElement(NormalizeConfig.FIELD_MAPPING, fieldMapping); failed = true; } if (!failed) { fieldList.add(mappings[0]); } } String[] fieldNormalizingArray = config.fieldNormalizing.split(","); String validTypeField = null; String validValueField = null; for (String fieldNormalizing : fieldNormalizingArray) { String[] fields = fieldNormalizing.split(":"); if (fields.length != 3 || Strings.isNullOrEmpty(fields[0]) || Strings.isNullOrEmpty(fields[1]) || Strings.isNullOrEmpty(fields[2])) { collector.addFailure("Normalizing field, field name, and field value must all be provided.", null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); continue; } if (inputSchema != null && inputSchema.getField(fields[0]) == null) { collector.addFailure( String.format("Normalizing field '%s' must be present in input schema.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (fieldList.contains(fields[0])) { collector.addFailure( String.format("Field '%s' cannot be used for both mapping and normalizing.", fields[0]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validTypeField == null) { validTypeField = fields[1]; } else if (!validTypeField.equals(fields[1])) { collector.addFailure( String.format("Invalid normalizing field type output column name '%s'.", fields[1]), "All normalizing field type output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (validValueField == null) { validValueField = fields[2]; } else if (!validValueField.equals(fields[2])) { collector.addFailure( String.format("Invalid normalizing field value output column name '%s'.", fields[2]), "All normalizing field value output column names must be the same.") .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[1]) == null) { collector.addFailure( String.format("Normalizing field type output column name '%s' must be present in output schema.", fields[1]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } if (outputSchema.getField(fields[2]) == null) { collector.addFailure( String.format("Normalizing field value output column name '%s' must be present in output schema.", fields[2]), null) .withConfigElement(NormalizeConfig.FIELD_NORMALIZING, fieldNormalizing); } } } Normalize(NormalizeConfig config); @Override void configurePipeline(PipelineConfigurer pipelineConfigurer); @Override void prepareRun(StageSubmitterContext context); @Override void initialize(TransformContext context); @Override void transform(StructuredRecord structuredRecord, Emitter<StructuredRecord> emitter);  }