@Override
	public void setConf(Configuration conf) {
	    this.conf = conf;
	    String ss  = conf.get("rhipe.hbase.value.type");
	    if(ss.equals("raw"))
		RHHBaseGeneral.valueType = 0;
	    else if(ss.equals("json"))
		RHHBaseGeneral.valueType = 1;
	    else RHHBaseGeneral.valueType = 0;
		
		String tableName = conf.get(INPUT_TABLE);
		try {
			setHTable(new HTable(HBaseConfiguration.create(conf), tableName));
		} catch (Exception e) {
			LOG.error(StringUtils.stringifyException(e));
		}
		
		Scan[] scans = null;
		if (conf.get(RHIPE_COLSPEC) != null) {
			try {
			    String[] cols = conf.get(RHIPE_COLSPEC).split(",");
			    CFQ = new byte[cols.length][][];
			    ArrayList<Pair<String,String>> l = null;
			    if(cols.length > 0){
				l = new ArrayList<Pair<String,String>>(cols.length);
				for(int i=0;i < cols.length;i++) {
				    CFQ[i] = new byte[2][];
				    String[] x = cols[i].split(":");
				    if(x.length==1){
					l.add(new Pair<String,String>(x[0],null));
					LOG.info("Added family: "+x[0]);
				    } else{
					l.add(new Pair<String,String>(x[0],x[1]));
					LOG.info("Added CF:Qualifier"+x[0]+":"+x[1]);
					CFQ[i][0] = x[0].getBytes();
					CFQ[i][1] = x[1].getBytes();
				    }
				}
			    }
			    String[] x = conf.get("rhipe.hbase.mozilla.cacheblocks").split(":");
			    scans = Util.generateScans(conf.get("rhipe.hbase.rowlim.start"),
						       conf.get("rhipe.hbase.rowlim.end"),
						       l,
						       Integer.parseInt(x[0]),
						       Integer.parseInt(x[1]) == 1? true: false);
			} catch (Exception e) {
				LOG.error("An error occurred.", e);
			}
		} else {
		    scans = new Scan[] { new Scan() };
		}
		
		setScans(scans);
	}