spark GBTClassifier 无法指定 rawPrediction这种情况如何计算auc？

谢邀！

AUC是分类模型常用的评价手段，目前的Spark mllib里面evaluation包中所提供的auc方法是拿到了roc曲线中的各个点之后再进行auc的计算，但是实际应用场景中（以逻辑回归为例），我们常常是对每个样本进行打分之后整合样本的label直接进行auc的计算，输入可能是（label, predict_score）这样的形式，mllib中提供的方案就不太适用了，所以这里提供了另一种计算方法，采用了针对0,1分类问题的近似计算方案，叫做BinaryAuc：

首先对predict_score进行排序，然后根据样本正负例的情况，分别计算每个小梯形的面积，最后汇总成为最终的auc值（由于在spark中数据是分布式RDD的形态，所以计算梯形面积的时候需要知道前一个RDD的offset，这里需要先遍历数据，但是避免了汇总到单机进行计算）：

1. 
   
   package org.apache.spark.mllib.wml
2. 
   
3. 
   /**
4. 
   * @author wangben 2015
5. 
   */
6. 
   import org.apache.spark.SparkContext
7. 
   import org.apache.spark.SparkConf
8. 
   import org.apache.spark.rdd.RDD
9. 
   import org.apache.spark.mllib.rdd.RDDFunctions._
10. 
    import scala.collection.Iterator
11. 
    import Array._
12. 
    
13. 
    class BinaryAUC extends Serializable {
14. 
    //input format: predictioin,label
15. 
    def auc( data: RDD[ (Double, Double) ] ) : Double =
16. 
    {
17. 
    //group same score result
18. 
    val group_result = data.groupByKey().map(x => {
19. 
    var r = new Array[Double](2)
20. 
    for(item <- x._2) {
21. 
    if(item > 0.0) r(1) += 1.0
22. 
    else r(0) += 1.0
23. 
    }
24. 
    (x._1, r) // score, [ FalseN, PositiveN ]
25. 
    })
26. 
    
27. 
    //points 需要累积
28. 
    val group_rank = group_result.sortByKey(false) //big first
29. 
    //计算累积
30. 
    var step_sizes = group_rank.mapPartitions( x =>
31. 
    {
32. 
    var r = List[(Double, Double)]()
33. 
    var fn_sum = 0.0
34. 
    var pn_sum = 0.0
35. 
    while( x.hasNext )
36. 
    {
37. 
    val cur = x.next
38. 
    fn_sum += cur._2(0)
39. 
    pn_sum += cur._2(1)
40. 
    }
41. 
    r.::(fn_sum, pn_sum).toIterator
42. 
    } ,true).collect
43. 
    var debug_string = ""
44. 
    var step_sizes_sum = ofDim[Double](step_sizes.size, 2) //二维数组
45. 
    for( i <- 0 to (step_sizes.size - 1) ) {
46. 
    if(i == 0) {
47. 
    step_sizes_sum(i)(0) = 0.0
48. 
    step_sizes_sum(i)(1) = 0.0
49. 
    } else {
50. 
    step_sizes_sum(i)(0) = step_sizes_sum(i - 1)(0) + step_sizes(i - 1)._1
51. 
    step_sizes_sum(i)(1) = step_sizes_sum(i - 1)(1) + step_sizes(i - 1)._2
52. 
    }
53. 
    debug_string += "\t" + step_sizes_sum(i)(0).toString + "\t" + step_sizes_sum(i)(1).toString
54. 
    }
55. 
    val sss_len = step_sizes_sum.size
56. 
    val total_fn = step_sizes_sum(sss_len - 1)(0) + step_sizes(sss_len - 1)._1
57. 
    val total_pn = step_sizes_sum(sss_len - 1)(1) + step_sizes(sss_len - 1)._2
58. 
    //System.out.println( "debug auc_step_size: " + debug_string)
59. 
    
60. 
    val bc_step_sizes_sum = data.context.broadcast(step_sizes_sum)
61. 
    val modified_group_rank = group_rank.mapPartitionsWithIndex( (index, x) =>
62. 
    {
63. 
    var sss = bc_step_sizes_sum.value
64. 
    var r = List[(Double, Array[Double])]()
65. 
    //var r = List[(Double, String)]()
66. 
    var fn = sss(index)(0) //start point
67. 
    var pn = sss(index)(1)
68. 
    while( x.hasNext )
69. 
    {
70. 
    var p = new Array[Double](2)
71. 
    val cur = x.next
72. 
    p(0) = fn + cur._2(0)
73. 
    p(1) = pn + cur._2(1)
74. 
    fn += cur._2(0)
75. 
    pn += cur._2(1)
76. 
    //r.::= (cur._1, p(0).toString() + "\t" + p(1).toString())
77. 
    r.::= (cur._1, p)
78. 
    }
79. 
    r.reverse.toIterator
80. 
    } ,true)
81. 
    
82. 
    //output debug info
83. 
    //modified_group_rank.map(l => l._1.toString + "\t" + l._2(0).toString + "\t" + l._2(1)).saveAsTextFile("/home/hdp_teu_dia/resultdata/wangben/debug_info")
84. 
    
85. 
    val score = modified_group_rank.sliding(2).aggregate(0.0)(
86. 
    seqOp = (auc: Double, points: Array[ (Double, Array[Double]) ]) => auc + TrapezoidArea(points),
87. 
    combOp = _ + _
88. 
    )
89. 
    System.out.println( "debug auc_mid: " + score
90. 
    + "\t" + (total_fn*total_pn).toString()
91. 
    + "\t" + total_fn.toString()
92. 
    + "\t" + total_pn.toString() )
93. 
    
94. 
    score/(total_fn*total_pn)
95. 
    }
96. 
    
97. 
    private def TrapezoidArea(points :Array[(Double, Array[Double])]):Double = {
98. 
    val x1 = points(0)._2(0)
99. 
    val y1 = points(0)._2(1)
100. 
     val x2 = points(1)._2(0)
101. 
     val y2 = points(1)._2(1)
102. 
     
103. 
     val base = x2 - x1
104. 
     val height = (y1 + y2)/2.0
105. 
     return base*height
106. 
     }
107. 
     }
108. 
     
109. 
     object AUCTest {
110. 
     
111. 
     def main(args: Array[String]){
112. 
     val conf=new SparkConf()
113. 
     conf.setAppName("TestEvaluation")
114. 
     val sc = new SparkContext(conf)
115. 
     val accum=sc.accumulator(0)
116. 
     val input_file = sc.textFile(args(0))
117. 
     val predict_label = input_file.map(l => {
118. 
     val x = l.stripPrefix("(").stripSuffix(")") split(",")
119. 
     (x(0).toDouble, x(1).toDouble)
120. 
     })
121. 
     val auc = new BinaryAUC()
122. 
     val auc_score = auc.auc(predict_label)
123. 
     System.out.println("debug auc_score: " + auc_score.toString())
124. 
     }
125. 
     
126. 
     }

按照这个方式试下！