@Override
	public void configure(JobConf conf) {
		
		 try {
			 FileSystem hdfs = FileSystem.get(conf);
			 FileUtil.copyMerge(hdfs, new Path(IOInfo.TEMP_MOVIE_OUTPUT), hdfs, new Path(IOInfo.TEMP_MOVIE_MERGE)
			 						, false, conf, null);
			 FileUtil.copyMerge(hdfs, new Path(IOInfo.TEMP_USER_OUTPUT), hdfs, new Path(IOInfo.TEMP_USER_MERGE)
			 						, false, conf, null);
		      String movieCacheName = new Path(IOInfo.TEMP_MOVIE_MERGE).getName();
		      String userCacheName = new Path(IOInfo.TEMP_USER_MERGE).getName();
		      Path [] cacheFiles = DistributedCache.getLocalCacheFiles(conf);
		      int count = 0;
		      if (null != cacheFiles && cacheFiles.length > 0) {
		        for (Path cachePath : cacheFiles) {
		        	if(cachePath.getName().equals(userCacheName)) {
		        		loadUserStat(cachePath);
		        		count++;
		        	} else if (cachePath.getName().equals(movieCacheName)) {
			            loadMovieStat(cachePath);
			            count++;
		        	}
		        	
		        	if(count>2) break;
		        }
		      }
		    } catch (IOException ioe) {
		      System.err.println("IOException reading from distributed cache");
		      System.err.println(ioe.toString());
		    }
		
	}