protected Schema generateSchemaWithNullifiedField ( WorkUnitState workUnitState , Schema currentAvroSchema ) { Configuration conf = new Configuration ( ) ; for ( String key : workUnitState . getPropertyNames ( ) ) { conf . set ( key , workUnitState . getProp ( key ) ) ; } // Get the original schema for merging. Path originalSchemaPath = null ; if ( workUnitState . contains ( CONVERTER_AVRO_NULLIFY_FIELDS_ORIGINAL_SCHEMA_PATH ) ) { originalSchemaPath = new Path ( workUnitState . getProp ( CONVERTER_AVRO_NULLIFY_FIELDS_ORIGINAL_SCHEMA_PATH ) ) ; } else { // If the path to get the original schema is not specified in the configuration, // adopt the best-try policy to search adjacent output folders. LOG . info ( "Property " + CONVERTER_AVRO_NULLIFY_FIELDS_ORIGINAL_SCHEMA_PATH + "is not specified. Trying to get the orignal schema from previous avro files." ) ; originalSchemaPath = WriterUtils . getDataPublisherFinalDir ( workUnitState , workUnitState . getPropAsInt ( ConfigurationKeys . FORK_BRANCHES_KEY , 1 ) , workUnitState . getPropAsInt ( ConfigurationKeys . FORK_BRANCH_ID_KEY , 0 ) ) . getParent ( ) ; } try { Schema prevSchema = AvroUtils . getDirectorySchema ( originalSchemaPath , conf , false ) ; Schema mergedSchema = AvroUtils . nullifyFieldsForSchemaMerge ( prevSchema , currentAvroSchema ) ; return mergedSchema ; } catch ( IOException ioe ) { LOG . error ( "Unable to nullify fields. Will retain the current avro schema." , ioe ) ; return currentAvroSchema ; } }
Generate new avro schema by nullifying fields that previously existed but not in the current schema .