public DataBag exec(Tuple input) {
        try {
            DataBag uiWeights = (DataBag) input.get(0);
            DataBag uiRecsTemp = (DataBag) input.get(1);

            //Create a set of items that the user has already seen.
            THashSet<String> seenBefore = new THashSet<String>();
            for (Tuple t : uiWeights) {
                    String item = (String) t.get(1);
                    seenBefore.add(item);
            }

            TObjectFloatHashMap<String> seenInRecs = new TObjectFloatHashMap<String>();
            HashMap<String, Tuple> candidates = new HashMap<String, Tuple>();

            //Go through the possible recommendations for the user and collect
            //the unseen recommendations.  Any item recommended for more than one
            //reason will only be saved once with the 'best' reason.
            for (Tuple t : uiRecsTemp) {
                Object item = t.get(1);
                // skip items already seen by user
                if (seenBefore.contains(item)) { continue; }

                // only take the best rec for an given item
                // out of any duplicate occurrences
                float weight = (Float) t.get(2);
                String sitem = (String) item;
                if ((!seenInRecs.containsKey(sitem))
                        || (weight > seenInRecs.get(sitem))) {
                        seenInRecs.put(sitem, weight);
                        candidates.put(sitem, t);
                }
            }

            DataBag outputBag = bf.newDefaultBag();
            List<Tuple> candValues = new ArrayList<Tuple>(candidates.values());

            if (diversityAdjust) {
                applyDiversityAdjustment(candValues);
            } else {
                // If we're not adjusting for diversity, the adjusted weight is just the current weight.
                for (Tuple t : candValues) {
                    t.append(t.get(2));
                }
            }
            
            //field @ pos 6 is diversity_adj_weight
            Collections.sort(candValues, getTupleComparator(6));
            
            for (int i = 0; i < Math.min(numRecs, candValues.size()); i++) {
                Tuple t = candValues.get(i);
                t.append(i + 1);
                outputBag.add(t);
            }

            return outputBag;
        } catch (ExecException e) {
            throw new RuntimeException(e);
        }
    }