[CRUNCH-657] Can't activate Snappy compression with AvroPathPerKeyTarget

• Type: Bug
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: 0.11.0
• Component/s: IO
• Environment: Cloudera
• Assignee: Unassigned
• Reporter: Jihed JOOBEUR
• Created: 25/Oct/17 14:19
• Updated: 28/Oct/17 04:19

Description

Compress.snappy(AvroPathPerKeyTarget) doesn't work.

I needed to create my own class that extends AvroPathPerKeyTarget.

public class CompressedAvroPathPerKeyTarget extends AvroPathPerKeyTarget {

    private Map<String, String> extraConf = Maps.newHashMap();

    public CompressedAvroPathPerKeyTarget(Path path) {
        super(path);
    }

    @Override
    public Target outputConf(String key, String value) {
        extraConf.put(key, value);
        return this;
    }

    @Override public void configureForMapReduce(Job job, PType<?> ptype, Path outputPath, String name) {
        AvroType<?> atype = (AvroType) ((PTableType) ptype).getValueType();
        FormatBundle bundle = FormatBundle.forOutput(AvroPathPerKeyOutputFormat.class);
        String schemaParam;
        if (name == null) {
            schemaParam = "avro.output.schema";
        } else {
            schemaParam = "avro.output.schema." + name;
        }

        for (Map.Entry<String, String> e : extraConf.entrySet()) {
            bundle.set(e.getKey(), e.getValue());
        }

        bundle.set(schemaParam, atype.getSchema().toString());
        AvroMode.fromType(atype).configure(bundle);
        configureForMapReduce(job, AvroWrapper.class, NullWritable.class, bundle, outputPath, name);
    }
}