@Override
		public void map(IntWritable key, IntArrayWritable value,
				OutputCollector<IntWritable, IntArrayWritable> output, Reporter reporter)
				throws IOException {
			IntWritable[] xMovie = (IntWritable[]) value.toArray();
			Set<Integer> imdbMovie = new HashSet<Integer>();
			for(int i =0 ;i<xMovie.length; i++) {
				imdbMovie.add(xMovie[i].get());
			}
			Map<Integer, Float> score = new HashMap<Integer, Float>();
			float minScore = Float.MAX_VALUE;
			for(Entry<Integer, Set<Integer>> e: nflixCluster.entrySet()){
				Set<Integer> intersection = new HashSet<Integer>(imdbMovie);
				Set<Integer> union = new HashSet<Integer> (imdbMovie);
				intersection.retainAll(e.getValue());
				float interSize = intersection.size();
				union.addAll(e.getValue());
				float unionSize = union.size();
				float dist= Float.MAX_VALUE;
				if(unionSize>0)
					dist = 1-(interSize/unionSize);
				if(dist<minScore) {
					minScore = dist;
				}
				score.put(e.getKey(), dist);
			}
			List<Integer> out = new ArrayList<Integer>();
			float threshold = Constant.CLUSTER_MAP_THRESHOLD*minScore;
			for(Entry<Integer, Float> e: score.entrySet()) {
				if(e.getValue()<=threshold) 
					out.add(new Integer(e.getKey()));
			}
			opValue.setIntegerCollection(out);
			output.collect(key, opValue);
		}