gistfile1.txt

package uk.co.danharvey.pig.storage;

import java.io.IOException;

public class JsonStorage extends StoreFunc {
	private RecordWriter<Text, NullWritable> writer;
	private ResourceSchema schema;
	
	@Override
	public OutputFormat getOutputFormat() throws IOException {
		return new TextOutputFormat<Text, NullWritable>();
	}
	
	@Override
	public void prepareToWrite(RecordWriter writer) throws IOException {
		this.writer = writer;
	}

	@Override
	public void putNext(Tuple tuple) throws IOException {
		JSONObject json = tupleToJson(null, tuple);
		try {
			writer.write(new Text(json.toString()), NullWritable.get());
			writer.write(new Text(s), NullWritable.get());
		} catch (InterruptedException e) {
			throw new IOException(e);
		}
	}
	
	@Override
	public void setStoreLocation(String location, Job job) throws IOException {
	    job.getConfiguration().set("mapred.textoutputformat.separator", "");
        FileOutputFormat.setOutputPath(job, new Path(location));
        if (location.endsWith(".bz2")) {
            FileOutputFormat.setCompressOutput(job, true);
            FileOutputFormat.setOutputCompressorClass(job,  BZip2Codec.class);
        }  else if (location.endsWith(".gz")) {
            FileOutputFormat.setCompressOutput(job, true);
            FileOutputFormat.setOutputCompressorClass(job, GzipCodec.class);
        }
	}
	
	@Override
	public void checkSchema(ResourceSchema s) {
		this.schema = s;
	}
}