public void initialize(InputSplit split, TaskAttemptContext arg1)
	throws IOException, InterruptedException {
		rows = new HashMap<ByteBuffer, List<ColumnOrSuperColumn>>();
		this.currentSplit = (CassandraRangeInputSplit) split;
		LOG.info("Starting retrieving split " + currentSplit);
		try {
			SlicePredicate slicePredicate = new SlicePredicate();
			SliceRange sliceRange = new SliceRange();
			sliceRange.setStart(new byte[] {});
			sliceRange.setFinish(new byte[] {});
			slicePredicate.setSlice_range(sliceRange);
			progress = 0;
			// TODO issue multiple queries to allow for really large splits
			KeyRange keyRange = new KeyRange(Constansts.MAX_SPLIT_SIZE)
				.setStart_token(currentSplit.getStartToken())
				.setEnd_token(currentSplit.getEndToken());
			ReliableCassandraClient client = new ReliableCassandraClient(currentSplit.getLocations(), currentSplit.getKeyspace());
			List<KeySlice> slices = client.get_range_slices(new ColumnParent(currentSplit.getColumnFamily()), slicePredicate, keyRange, ConsistencyLevel.ONE);
			for(KeySlice k : slices){
				rows.put(k.key, k.columns);
			}
			keyIterator = rows.keySet().iterator();
			LOG.error("Got " + rows.size() + " rows.");
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw e;
		}
		//}
	}