A Crunch MapFn that can project Avro generic record fields

AvroProjectionMapFn.java

public class AvroProjectionMapFn extends MapFn<Record, Record> {

    private static final long serialVersionUID = 1L;

    private List<String> fieldNames;
    private Schema sourceSchema;

    private Schema projectionSchema;

    @Override
    public void initialize() {
      List<Schema.Field> destFields = Lists.newArrayList();

      for (String fieldName : fieldNames) {
        Schema.Field sourceField = sourceSchema.getField(fieldName);
        destFields.add(sourceField);
      }

      projectionSchema = Schema.createRecord(destFields);

      super.initialize();
    }

    @Override
    public Record map(Record input) {
      Record result = new Record(projectionSchema);

      for (Schema.Field field : projectionSchema.getFields()) {
        result.put(field.name(), input.get(field.name()));
      }

      return result;
    }

    public Schema getProjectionSchema() {
      return projectionSchema;
    }

    public Schema getSourceSchema() {
      return sourceSchema;
    }

    public void setSourceSchema(Schema sourceSchema) {
      this.sourceSchema = sourceSchema;
    }

    public void setFieldNames(String... fieldNames) {
      this.fieldNames = Lists.newArrayList(fieldNames);
    }

}