Details
-
Bug
-
Status: Open
-
Major
-
Resolution: Unresolved
-
0.11.0
-
None
-
None
-
Cloudera
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); } }