@@ -0,0 +1,2605 @@
+package pg.hadoop.pig.piggybank;%0A%0Aimport org.apache.hadoop.mapreduce.RecordWriter;%0Aimport org.apache.hadoop.mapreduce.TaskAttemptContext;%0Aimport org.apache.pig.data.Tuple;%0Aimport org.apache.pig.data.TupleFactory;%0Aimport org.apache.pig.piggybank.storage.MultiStorage;%0A%0Aimport java.io.IOException;%0Aimport java.util.List;%0A%0A/**%0A * Extended MultiStorage class that allows to remove the key field from the outputted files.%0A */%0Apublic class ExtendedMultiStorage extends MultiStorage %7B%0A%0A%0A    protected static final String DEFAULT_COMPRESSION = %22none%22;%0A    protected static final String DEFAULT_DELIMITER = %22%5C%5Ct%22;%0A%0A    protected final int keyIndex;%0A    protected final boolean includeFieldIndex;%0A%0A    public ExtendedMultiStorage(String parentPathStr, String splitFieldIndex, String includeFieldIndex) %7B%0A        this(parentPathStr, splitFieldIndex, DEFAULT_COMPRESSION, includeFieldIndex);%0A    %7D%0A%0A    public ExtendedMultiStorage(String parentPathStr, String splitFieldIndex, String compression, String includeFieldIndex) %7B%0A        this(parentPathStr, splitFieldIndex, compression, DEFAULT_DELIMITER, includeFieldIndex);%0A    %7D%0A%0A    public ExtendedMultiStorage(String parentPathStr, String splitFieldIndex, String compression, String fieldDel, String includeFieldIndex) %7B%0A        super(parentPathStr, splitFieldIndex, compression, fieldDel);%0A        this.keyIndex = Integer.parseInt(splitFieldIndex);%0A        this.includeFieldIndex = Boolean.parseBoolean(includeFieldIndex);%0A    %7D%0A%0A    @Override%0A    public void prepareToWrite(RecordWriter writer) throws IOException %7B%0A        super.prepareToWrite(includeFieldIndex ? writer : new RecordWriterWrapper(writer));%0A    %7D%0A%0A    /**%0A     * Simple wrapper that intercepts the write call and remove the unnecessary field from the tuple before the write.%0A     */%0A    private class RecordWriterWrapper extends RecordWriter%3CString, Tuple%3E %7B%0A%0A        private RecordWriter%3CString, Tuple%3E originalWriter;%0A%0A        private RecordWriterWrapper(final RecordWriter%3CString, Tuple%3E originalWriter) %7B%0A            this.originalWriter = originalWriter;%0A        %7D%0A%0A        @Override%0A        public void write(String key, Tuple value) throws IOException, InterruptedException %7B%0A            final List%3CObject%3E fields = value.getAll();%0A            fields.remove(keyIndex);%0A            final Tuple newTuple = TupleFactory.getInstance().newTupleNoCopy(fields);%0A            originalWriter.write(key, newTuple);%0A        %7D%0A%0A        @Override%0A        public void close(TaskAttemptContext context) throws IOException, InterruptedException %7B%0A            originalWriter.close(context);%0A        %7D%0A    %7D%0A%7D%0A

Add ExtendedMultiStorage.