app/oryx-app-mllib/src/main/java/com/cloudera/oryx/app/batch/mllib/als/Evaluation.java
static double areaUnderCurve(JavaSparkContext sparkContext,
                               MatrixFactorizationModel mfModel,
                               JavaRDD<Rating> positiveData) {

    // This does not use Spark's BinaryClassificationMetrics.areaUnderROC because it
    // is intended to operate on one large set of (score,label) pairs. The computation
    // here is really many small AUC problems, for which a much faster direct computation
    // is available.

    // Extract all positive (user,product) pairs
    JavaPairRDD<Integer,Integer> positiveUserProducts =
        positiveData.mapToPair(rating -> new Tuple2<>(rating.user(), rating.product()));

    JavaPairRDD<Integer,Iterable<Rating>> positivePredictions =
        predictAll(mfModel, positiveData, positiveUserProducts);

    // All distinct item IDs, to be broadcast
    Broadcast<List<Integer>> allItemIDsBC = sparkContext.broadcast(positiveUserProducts.values().distinct().collect());

    JavaPairRDD<Integer,Integer> negativeUserProducts =
      positiveUserProducts.groupByKey().flatMapToPair(
        new PairFlatMapFunction<Tuple2<Integer,Iterable<Integer>>,Integer,Integer>() {
          private final RandomGenerator random = RandomManager.getRandom();
          @Override
          public Iterator<Tuple2<Integer,Integer>> call(
              Tuple2<Integer,Iterable<Integer>> userIDsAndItemIDs) {
            Integer userID = userIDsAndItemIDs._1();
            Collection<Integer> positiveItemIDs = Sets.newHashSet(userIDsAndItemIDs._2());
            int numPositive = positiveItemIDs.size();
            Collection<Tuple2<Integer,Integer>> negative = new ArrayList<>(numPositive);
            List<Integer> allItemIDs = allItemIDsBC.value();
            int numItems = allItemIDs.size();
            // Sample about as many negative examples as positive
            for (int i = 0; i < numItems && negative.size() < numPositive; i++) {
              Integer itemID = allItemIDs.get(random.nextInt(numItems));
              if (!positiveItemIDs.contains(itemID)) {
                negative.add(new Tuple2<>(userID, itemID));
              }
            }
            return negative.iterator();
          }
        });

    JavaPairRDD<Integer,Iterable<Rating>> negativePredictions =
        predictAll(mfModel, positiveData, negativeUserProducts);

    return positivePredictions.join(negativePredictions).values().mapToDouble(t -> {
        // AUC is also the probability that random positive examples
        // rank higher than random examples at large. Here we compare all random negative
        // examples to all positive examples and report the totals as an alternative
        // computation for AUC
        long correct = 0;
        long total = 0;
        for (Rating positive : t._1()) {
          for (Rating negative : t._2()) {
            if (positive.rating() > negative.rating()) {
              correct++;
            }
            total++;
          }
        }
        if (total == 0) {
          return 0.0;
        }
        return (double) correct / total;
      }).mean();
  }
static double areaUnderCurve(JavaSparkContext sparkContext,
                               MatrixFactorizationModel mfModel,
                               JavaRDD<Rating> positiveData) {

    // This does not use Spark's BinaryClassificationMetrics.areaUnderROC because it
    // is intended to operate on one large set of (score,label) pairs. The computation
    // here is really many small AUC problems, for which a much faster direct computation
    // is available.

    // Extract all positive (user,product) pairs
    JavaPairRDD<Integer,Integer> positiveUserProducts =
        positiveData.mapToPair(rating -> new Tuple2<>(rating.user(), rating.product()));

    JavaPairRDD<Integer,Iterable<Rating>> positivePredictions =
        predictAll(mfModel, positiveData, positiveUserProducts);

    // All distinct item IDs, to be broadcast
    Broadcast<List<Integer>> allItemIDsBC = sparkContext.broadcast(positiveUserProducts.values().distinct().collect());

    JavaPairRDD<Integer,Integer> negativeUserProducts =
      positiveUserProducts.groupByKey().flatMapToPair(
        new PairFlatMapFunction<Tuple2<Integer,Iterable<Integer>>,Integer,Integer>() {
          private final RandomGenerator random = RandomManager.getRandom();
          @Override
          public Iterator<Tuple2<Integer,Integer>> call(
              Tuple2<Integer,Iterable<Integer>> userIDsAndItemIDs) {
            Integer userID = userIDsAndItemIDs._1();
            Collection<Integer> positiveItemIDs = Sets.newHashSet(userIDsAndItemIDs._2());
            int numPositive = positiveItemIDs.size();
            Collection<Tuple2<Integer,Integer>> negative = new ArrayList<>(numPositive);
            List<Integer> allItemIDs = allItemIDsBC.value();
            int numItems = allItemIDs.size();
            // Sample about as many negative examples as positive
            for (int i = 0; i < numItems && negative.size() < numPositive; i++) {
              Integer itemID = allItemIDs.get(random.nextInt(numItems));
              if (!positiveItemIDs.contains(itemID)) {
                negative.add(new Tuple2<>(userID, itemID));
              }
            }
            return negative.iterator();
          }
        });

    JavaPairRDD<Integer,Iterable<Rating>> negativePredictions =
        predictAll(mfModel, positiveData, negativeUserProducts);

    return positivePredictions.join(negativePredictions).values().mapToDouble(t -> {
        // AUC is also the probability that random positive examples
        // rank higher than random examples at large. Here we compare all random negative
        // examples to all positive examples and report the totals as an alternative
        // computation for AUC
        long correct = 0;
        long total = 0;
        for (Rating positive : t._1()) {
          for (Rating negative : t._2()) {
            if (positive.rating() > negative.rating()) {
              correct++;
            }
            total++;
          }
        }
        if (total == 0) {
          return 0.0;
        }
        return (double) correct / total;
      }).mean();
  }
static double areaUnderCurve ( JavaSparkContext sparkContext , MatrixFactorizationModel mfModel , JavaRDD < Rating > positiveData ) { // This does not use Spark's BinaryClassificationMetrics.areaUnderROC because it // is intended to operate on one large set of (score,label) pairs. The computation // here is really many small AUC problems, for which a much faster direct computation // is available. // Extract all positive (user,product) pairs JavaPairRDD < Integer , Integer > positiveUserProducts = positiveData . mapToPair ( rating -> new Tuple2 <> ( rating . user ( ) , rating . product ( ) ) ) ; JavaPairRDD < Integer , Iterable < Rating > > positivePredictions = predictAll ( mfModel , positiveData , positiveUserProducts ) ; // All distinct item IDs, to be broadcast Broadcast < List < Integer > > allItemIDsBC = sparkContext . broadcast ( positiveUserProducts . values ( ) . distinct ( ) . collect ( ) ) ; JavaPairRDD < Integer , Integer > negativeUserProducts = positiveUserProducts . groupByKey ( ) . flatMapToPair ( new PairFlatMapFunction < Tuple2 < Integer , Iterable < Integer > > , Integer , Integer > ( ) { private final RandomGenerator random = RandomManager . getRandom ( ) ; @ Override public Iterator < Tuple2 < Integer , Integer > > call ( Tuple2 < Integer , Iterable < Integer > > userIDsAndItemIDs ) { Integer userID = userIDsAndItemIDs . _1 ( ) ; Collection < Integer > positiveItemIDs = Sets . newHashSet ( userIDsAndItemIDs . _2 ( ) ) ; int numPositive = positiveItemIDs . size ( ) ; Collection < Tuple2 < Integer , Integer > > negative = new ArrayList <> ( numPositive ) ; List < Integer > allItemIDs = allItemIDsBC . value ( ) ; int numItems = allItemIDs . size ( ) ; // Sample about as many negative examples as positive for ( int i = 0 ; i < numItems && negative . size ( ) < numPositive ; i ++ ) { Integer itemID = allItemIDs . get ( random . nextInt ( numItems ) ) ; if ( ! positiveItemIDs . contains ( itemID ) ) { negative . add ( new Tuple2 <> ( userID , itemID ) ) ; } } return negative . iterator ( ) ; } } ) ; JavaPairRDD < Integer , Iterable < Rating > > negativePredictions = predictAll ( mfModel , positiveData , negativeUserProducts ) ; return positivePredictions . join ( negativePredictions ) . values ( ) . mapToDouble ( t -> { // AUC is also the probability that random positive examples // rank higher than random examples at large. Here we compare all random negative // examples to all positive examples and report the totals as an alternative // computation for AUC long correct = 0 ; long total = 0 ; for ( Rating positive : t . _1 ( ) ) { for ( Rating negative : t . _2 ( ) ) { if ( positive . rating ( ) > negative . rating ( ) ) { correct ++ ; } total ++ ; } } if ( total == 0 ) { return 0.0 ; } return ( double ) correct / total ; } ) . mean ( ) ; }
Computes AUC (area under the ROC curve) as a recommender evaluation metric.
Really, it computes what might be described as "Mean AUC", as it computes AUC per
user and averages them.
Computes AUC ( area under the ROC curve ) as a recommender evaluation metric . Really it computes what might be described as Mean AUC as it computes AUC per user and averages them .