<commit_before>package hipi.experiments.mapreduce;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import hipi.image.FloatImage;
import hipi.image.ImageHeader;
import hipi.image.ImageHeader.ImageType;
import hipi.image.io.CodecManager;
import hipi.image.io.ImageDecoder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class JPEGSequenceFileRecordReader
extends RecordReader<ImageHeader, FloatImage> {
	private SequenceFile.Reader _reader;
	private byte _cacheData[];
	private int _cacheType;
	private long _start;
	private long _end;

	@Override
	public void initialize(InputSplit isplit, TaskAttemptContext job)
	throws IOException, InterruptedException {
		FileSplit split = (FileSplit) isplit;
		Path path = split.getPath();

		Configuration conf = job.getConfiguration();
		FileSystem fs = FileSystem.get(conf);
		_reader = new SequenceFile.Reader(fs, path, conf);
		_start = split.getStart();
		_end = split.getStart() + split.getLength();
		if (split.getStart() > _reader.getPosition())
			_reader.sync(split.getStart());   

		_cacheData = null;
		_cacheType = 0;		
	}


	@Override
	public ImageHeader getCurrentKey() throws IOException, InterruptedException {
		if (_cacheData != null) {
			ImageDecoder decoder = CodecManager.getDecoder(ImageType.fromValue(_cacheType));
			if (decoder == null)
				return null;
			ByteArrayInputStream bis = new ByteArrayInputStream(_cacheData);
			ImageHeader header = decoder.decodeImageHeader(bis);
			bis.close();
			return header;
		}
		return null;
	}

	@Override
	public FloatImage getCurrentValue() throws IOException,
	InterruptedException {
		if (_cacheData != null) {
			ImageDecoder decoder = CodecManager.getDecoder(ImageType.fromValue(_cacheType));
			if (decoder == null)
				return null;
			ByteArrayInputStream bis = new ByteArrayInputStream(_cacheData);
			FloatImage image = decoder.decodeImage(bis);
			bis.close();
			return image;
		}
		return null;
	}


	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if(_reader.getPosition() > _end){
			return false;			
		}
		try {
			LongWritable key = new LongWritable();
			BytesWritable data = new BytesWritable();<commit_msg>modified JPEGSequenceFileRecordReader to be more robust to failure<commit_after>package hipi.experiments.mapreduce;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import hipi.image.FloatImage;
import hipi.image.ImageHeader;
import hipi.image.ImageHeader.ImageType;
import hipi.image.io.CodecManager;
import hipi.image.io.ImageDecoder;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class JPEGSequenceFileRecordReader
extends RecordReader<ImageHeader, FloatImage> {
	private SequenceFile.Reader _reader;
	private byte _cacheData[];
	private int _cacheType;
	private long _start;
	private long _end;

	@Override
	public void initialize(InputSplit isplit, TaskAttemptContext job)
	throws IOException, InterruptedException {
		FileSplit split = (FileSplit) isplit;
		Path path = split.getPath();

		Configuration conf = job.getConfiguration();
		FileSystem fs = FileSystem.get(conf);
		_reader = new SequenceFile.Reader(fs, path, conf);
		_start = split.getStart();
		_end = split.getStart() + split.getLength();
		if (split.getStart() > _reader.getPosition())
			_reader.sync(split.getStart());   

		_cacheData = null;
		_cacheType = 0;		
	}


	@Override
	public ImageHeader getCurrentKey() throws IOException, InterruptedException {
		if (_cacheData != null) {
			ImageDecoder decoder = CodecManager.getDecoder(ImageType.fromValue(_cacheType));
			if (decoder == null)
				return null;
			try {
				ByteArrayInputStream bis = new ByteArrayInputStream(_cacheData);
				ImageHeader header = decoder.decodeImageHeader(bis);
				bis.close();
				return header;
			} catch (Exception e) {
				return null;
			}
		}
		return null;
	}

	@Override
	public FloatImage getCurrentValue() throws IOException,
	InterruptedException {
		if (_cacheData != null) {
			ImageDecoder decoder = CodecManager.getDecoder(ImageType.fromValue(_cacheType));
			if (decoder == null)
				return null;
			try {
				ByteArrayInputStream bis = new ByteArrayInputStream(_cacheData);
				FloatImage image = decoder.decodeImage(bis);
				bis.close();
				return image;
			} catch (Exception e) {
				return null;
			}
		}
		return null;
	}


	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if(_reader.getPosition() > _end){
			return false;			
		}
		try {
			LongWritable key = new LongWritable();
			BytesWritable data = new BytesWritable();
{'fraction_non_alphanumeric': 0.10072477234714737, 'fraction_numerical': 0.00037167812674224123, 'mean_word_length': 4.888402625820569, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1920177', 'n_tokens_mistral': 1769, 'n_tokens_neox': 1596, 'n_words': 374}