@Test public void testSwitchNamespace() { String originalNamespace = "originalNamespace"; String originalName = "originalName"; String newNamespace = "newNamespace"; Schema schema = SchemaBuilder.builder(originalNamespace).record(originalName).fields(). requiredDouble("double").optionalFloat("float").endRecord(); Map<String, String> map = Maps.newHashMap(); map.put(originalNamespace, newNamespace); Schema newSchema = AvroUtils.switchNamespace(schema, map); Assert.assertEquals(newSchema.getNamespace(), newNamespace); Assert.assertEquals(newSchema.getName(), originalName); for(Schema.Field field : newSchema.getFields()) { Assert.assertEquals(field, schema.getField(field.name())); } }
public static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride) { Schema newSchema; String newNamespace = StringUtils.EMPTY; switch (schema.getType()) { case ENUM: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createEnum(schema.getName(), schema.getDoc(), newNamespace, schema.getEnumSymbols()); break; case FIXED: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createFixed(schema.getName(), schema.getDoc(), newNamespace, schema.getFixedSize()); break; case MAP: newSchema = Schema.createMap(switchNamespace(schema.getValueType(), namespaceOverride)); break; case RECORD: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); List<Schema.Field> newFields = new ArrayList<>(); if (schema.getFields().size() > 0) { for (Schema.Field oldField : schema.getFields()) { Field newField = new Field(oldField.name(), switchNamespace(oldField.schema(), namespaceOverride), oldField.doc(), oldField.defaultValue(), oldField.order()); newFields.add(newField); } } newSchema = Schema.createRecord(schema.getName(), schema.getDoc(), newNamespace, schema.isError()); newSchema.setFields(newFields); break; case UNION: List<Schema> newUnionMembers = new ArrayList<>(); if (null != schema.getTypes() && schema.getTypes().size() > 0) { for (Schema oldUnionMember : schema.getTypes()) { newUnionMembers.add(switchNamespace(oldUnionMember, namespaceOverride)); } } newSchema = Schema.createUnion(newUnionMembers); break; case ARRAY: newSchema = Schema.createArray(switchNamespace(schema.getElementType(), namespaceOverride)); break; case BOOLEAN: case BYTES: case DOUBLE: case FLOAT: case INT: case LONG: case NULL: case STRING: newSchema = Schema.create(schema.getType()); break; default: String exceptionMessage = String.format("Schema namespace replacement failed for \"%s\" ", schema); LOG.error(exceptionMessage); throw new AvroRuntimeException(exceptionMessage); } copyProperties(schema, newSchema); return newSchema; }
AvroUtils { public static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride) { Schema newSchema; String newNamespace = StringUtils.EMPTY; switch (schema.getType()) { case ENUM: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createEnum(schema.getName(), schema.getDoc(), newNamespace, schema.getEnumSymbols()); break; case FIXED: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createFixed(schema.getName(), schema.getDoc(), newNamespace, schema.getFixedSize()); break; case MAP: newSchema = Schema.createMap(switchNamespace(schema.getValueType(), namespaceOverride)); break; case RECORD: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); List<Schema.Field> newFields = new ArrayList<>(); if (schema.getFields().size() > 0) { for (Schema.Field oldField : schema.getFields()) { Field newField = new Field(oldField.name(), switchNamespace(oldField.schema(), namespaceOverride), oldField.doc(), oldField.defaultValue(), oldField.order()); newFields.add(newField); } } newSchema = Schema.createRecord(schema.getName(), schema.getDoc(), newNamespace, schema.isError()); newSchema.setFields(newFields); break; case UNION: List<Schema> newUnionMembers = new ArrayList<>(); if (null != schema.getTypes() && schema.getTypes().size() > 0) { for (Schema oldUnionMember : schema.getTypes()) { newUnionMembers.add(switchNamespace(oldUnionMember, namespaceOverride)); } } newSchema = Schema.createUnion(newUnionMembers); break; case ARRAY: newSchema = Schema.createArray(switchNamespace(schema.getElementType(), namespaceOverride)); break; case BOOLEAN: case BYTES: case DOUBLE: case FLOAT: case INT: case LONG: case NULL: case STRING: newSchema = Schema.create(schema.getType()); break; default: String exceptionMessage = String.format("Schema namespace replacement failed for \"%s\" ", schema); LOG.error(exceptionMessage); throw new AvroRuntimeException(exceptionMessage); } copyProperties(schema, newSchema); return newSchema; } }
AvroUtils { public static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride) { Schema newSchema; String newNamespace = StringUtils.EMPTY; switch (schema.getType()) { case ENUM: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createEnum(schema.getName(), schema.getDoc(), newNamespace, schema.getEnumSymbols()); break; case FIXED: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createFixed(schema.getName(), schema.getDoc(), newNamespace, schema.getFixedSize()); break; case MAP: newSchema = Schema.createMap(switchNamespace(schema.getValueType(), namespaceOverride)); break; case RECORD: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); List<Schema.Field> newFields = new ArrayList<>(); if (schema.getFields().size() > 0) { for (Schema.Field oldField : schema.getFields()) { Field newField = new Field(oldField.name(), switchNamespace(oldField.schema(), namespaceOverride), oldField.doc(), oldField.defaultValue(), oldField.order()); newFields.add(newField); } } newSchema = Schema.createRecord(schema.getName(), schema.getDoc(), newNamespace, schema.isError()); newSchema.setFields(newFields); break; case UNION: List<Schema> newUnionMembers = new ArrayList<>(); if (null != schema.getTypes() && schema.getTypes().size() > 0) { for (Schema oldUnionMember : schema.getTypes()) { newUnionMembers.add(switchNamespace(oldUnionMember, namespaceOverride)); } } newSchema = Schema.createUnion(newUnionMembers); break; case ARRAY: newSchema = Schema.createArray(switchNamespace(schema.getElementType(), namespaceOverride)); break; case BOOLEAN: case BYTES: case DOUBLE: case FLOAT: case INT: case LONG: case NULL: case STRING: newSchema = Schema.create(schema.getType()); break; default: String exceptionMessage = String.format("Schema namespace replacement failed for \"%s\" ", schema); LOG.error(exceptionMessage); throw new AvroRuntimeException(exceptionMessage); } copyProperties(schema, newSchema); return newSchema; }  }
AvroUtils { public static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride) { Schema newSchema; String newNamespace = StringUtils.EMPTY; switch (schema.getType()) { case ENUM: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createEnum(schema.getName(), schema.getDoc(), newNamespace, schema.getEnumSymbols()); break; case FIXED: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createFixed(schema.getName(), schema.getDoc(), newNamespace, schema.getFixedSize()); break; case MAP: newSchema = Schema.createMap(switchNamespace(schema.getValueType(), namespaceOverride)); break; case RECORD: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); List<Schema.Field> newFields = new ArrayList<>(); if (schema.getFields().size() > 0) { for (Schema.Field oldField : schema.getFields()) { Field newField = new Field(oldField.name(), switchNamespace(oldField.schema(), namespaceOverride), oldField.doc(), oldField.defaultValue(), oldField.order()); newFields.add(newField); } } newSchema = Schema.createRecord(schema.getName(), schema.getDoc(), newNamespace, schema.isError()); newSchema.setFields(newFields); break; case UNION: List<Schema> newUnionMembers = new ArrayList<>(); if (null != schema.getTypes() && schema.getTypes().size() > 0) { for (Schema oldUnionMember : schema.getTypes()) { newUnionMembers.add(switchNamespace(oldUnionMember, namespaceOverride)); } } newSchema = Schema.createUnion(newUnionMembers); break; case ARRAY: newSchema = Schema.createArray(switchNamespace(schema.getElementType(), namespaceOverride)); break; case BOOLEAN: case BYTES: case DOUBLE: case FLOAT: case INT: case LONG: case NULL: case STRING: newSchema = Schema.create(schema.getType()); break; default: String exceptionMessage = String.format("Schema namespace replacement failed for \"%s\" ", schema); LOG.error(exceptionMessage); throw new AvroRuntimeException(exceptionMessage); } copyProperties(schema, newSchema); return newSchema; }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); }
AvroUtils { public static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride) { Schema newSchema; String newNamespace = StringUtils.EMPTY; switch (schema.getType()) { case ENUM: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createEnum(schema.getName(), schema.getDoc(), newNamespace, schema.getEnumSymbols()); break; case FIXED: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); newSchema = Schema.createFixed(schema.getName(), schema.getDoc(), newNamespace, schema.getFixedSize()); break; case MAP: newSchema = Schema.createMap(switchNamespace(schema.getValueType(), namespaceOverride)); break; case RECORD: newNamespace = namespaceOverride.containsKey(schema.getNamespace()) ? namespaceOverride.get(schema.getNamespace()) : schema.getNamespace(); List<Schema.Field> newFields = new ArrayList<>(); if (schema.getFields().size() > 0) { for (Schema.Field oldField : schema.getFields()) { Field newField = new Field(oldField.name(), switchNamespace(oldField.schema(), namespaceOverride), oldField.doc(), oldField.defaultValue(), oldField.order()); newFields.add(newField); } } newSchema = Schema.createRecord(schema.getName(), schema.getDoc(), newNamespace, schema.isError()); newSchema.setFields(newFields); break; case UNION: List<Schema> newUnionMembers = new ArrayList<>(); if (null != schema.getTypes() && schema.getTypes().size() > 0) { for (Schema oldUnionMember : schema.getTypes()) { newUnionMembers.add(switchNamespace(oldUnionMember, namespaceOverride)); } } newSchema = Schema.createUnion(newUnionMembers); break; case ARRAY: newSchema = Schema.createArray(switchNamespace(schema.getElementType(), namespaceOverride)); break; case BOOLEAN: case BYTES: case DOUBLE: case FLOAT: case INT: case LONG: case NULL: case STRING: newSchema = Schema.create(schema.getType()); break; default: String exceptionMessage = String.format("Schema namespace replacement failed for \"%s\" ", schema); LOG.error(exceptionMessage); throw new AvroRuntimeException(exceptionMessage); } copyProperties(schema, newSchema); return newSchema; }  static boolean checkReaderWriterCompatibility(Schema readerSchema, Schema writerSchema, boolean ignoreNamespace); static Schema addSchemaCreationTime(Schema inputSchema, Schema outputSchema); static String getSchemaCreationTime(Schema inputSchema); static Schema setSchemaCreationTime(Schema inputSchema, String creationTime); static List<Field> deepCopySchemaFields(Schema readerSchema); static void convertFieldToSchemaWithProps(Map<String,JsonNode> fieldProps, Schema targetSchemaObj); static Optional<Schema> getFieldSchema(Schema schema, String fieldLocation); static Optional<Field> getField(Schema schema, String fieldLocation); static Optional<Object> getFieldValue(GenericRecord record, String fieldLocation); static Map<String, Object> getMultiFieldValue(GenericRecord record, String fieldLocation); @SuppressWarnings("unchecked") static Map<String, String> toStringMap(Object map); static GenericRecord convertRecordSchema(GenericRecord record, Schema newSchema); static byte[] recordToByteArray(GenericRecord record); static Schema getSchemaFromDataFile(Path dataFile, FileSystem fs); static Schema parseSchemaFromFile(Path filePath, FileSystem fs); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite); static void writeSchemaToFile(Schema schema, Path filePath, FileSystem fs, boolean overwrite, FsPermission perm); static void writeSchemaToFile(Schema schema, Path filePath, Path tempFilePath, FileSystem fs, boolean overwrite,
      FsPermission perm); static Schema getDirectorySchema(Path directory, FileSystem fs, boolean latest); static Schema getDirectorySchema(Path directory, Configuration conf, boolean latest); static Schema nullifyFieldsForSchemaMerge(Schema oldSchema, Schema newSchema); static Optional<Schema> removeUncomparableFields(Schema schema); static Schema switchName(Schema schema, String newName); static Schema switchNamespace(Schema schema, Map<String, String> namespaceOverride); static Path serializeAsPath(GenericRecord record, boolean includeFieldNames, boolean replacePathSeparators); static String sanitizeSchemaString(String schemaString); static GenericRecord slowDeserializeGenericRecord(byte[] serializedRecord, Schema schema); static Schema decorateRecordSchema(Schema inputSchema, @Nonnull List<Field> fieldList); static GenericRecord decorateRecord(GenericRecord inputRecord, @Nonnull Map<String, Object> fieldMap,
      Schema outputSchema); static GenericRecord overrideNameAndNamespace(GenericRecord input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static Schema overrideNameAndNamespace(Schema input, String nameOverride, Optional<Map<String, String>> namespaceOverride); static boolean isSchemaRecursive(Schema schema, Optional<Logger> logger); static Pair<Schema, List<SchemaEntry>> dropRecursiveFields(Schema schema); static final String FIELD_LOCATION_DELIMITER; }