Class PigStreamingUDF

src/org/apache/pig/impl/streaming/PigStreamingUDF.java:47–224

import com.google.common.base.Charsets;

public class PigStreamingUDF extends PigStreamingBase {
    private static final byte PRE_WRAP_DELIM = '|';
    private static final byte POST_WRAP_DELIM = '_';
    private static final StreamingDelimiters DELIMS = 
            new StreamingDelimiters(PRE_WRAP_DELIM, POST_WRAP_DELIM, false);
    
    private FieldSchema topLevelFs;
    private static TupleFactory tupleFactory = TupleFactory.getInstance();
    private static BagFactory bagFactory = BagFactory.getInstance();

    private WritableByteArray out;

    public PigStreamingUDF() {
        out = new WritableByteArray();
    }
    
    public PigStreamingUDF(FieldSchema topLevelFs) {
        out = new WritableByteArray();
        this.topLevelFs = topLevelFs;
    }

    @Override
    public WritableByteArray serializeToBytes(Tuple t) throws IOException {
        out.reset();
        int sz;
        Object field;
        if (t == null) {
            sz = 0;
        } else {
            sz = t.size();
        }
        for (int i=0; i < sz; i++) {
            field = t.get(i);
            StorageUtil.putField(out, field, DELIMS, true);
            if (i != sz-1) {
                out.write(DELIMS.getParamDelim());
            }
        }
        byte[] recordDel = DELIMS.getRecordEnd();
        out.write(recordDel, 0, recordDel.length);
        return out;
    }
    
    @Override
    public LoadCaster getLoadCaster() throws IOException {
        return new Utf8StorageConverter();
    }
    
    @Override
    public Tuple deserialize(byte[] bytes, int offset, int length) throws IOException {
        Object o = deserialize(topLevelFs, bytes, 0 + offset, length - DELIMS.getRecordEnd().length); //Drop newline
        return tupleFactory.newTuple(o);
    }
    
    public byte[] getRecordDelim() {
        return DELIMS.getRecordEnd();
    }