XGBoost缺失值引发的问题及其深度分析 | CSDN博文精选

  scala
  /**
   * @return A tuple of the booster and the metrics used to build training summary
   */
  @throws(classOf[XGBoostError])
  def trainDistributed(
      trainingDataIn: RDD[XGBLabeledPoint],
      params: Map[String, Any],
      round: Int,
      nWorkers: Int,
      obj: ObjectiveTrait = null,
      eval: EvalTrait = null,
      useExternalMemory: Boolean = false




    
,

      //NaN作为missing的值
      missing: Float = Float.NaN,

      hasGroup: Boolean = false): (Booster, Map[String, Array[Float]]) = {
      //...
      }

    //测试结果中的一行，41列
    double[] input = new double[]{1, 2, 5, 0, 0, 6.666666666666667, 31.14, 29.28, 0, 1.303333, 2.8555, 2.37, 701, 463, 3.989, 3.85, 14400.5, 15.79, 11.45, 0.915, 7.05, 5.5, 0.023333, 0.0365, 0.0275, 0.123333, 0.4645, 0.12, 15.082, 14.48, 0, 31.8425, 29.1, 7.7325, 3, 5.88, 1.08, 0, 0, 0, 32];
    float[] testInput = new float[input.length];
    for(int i = 0, total = input.length; i 
      testInput[i] = new Double(input[i]).floatValue();
    }
    Booster booster = XGBoost.loadModel("${model}");
    //一行，41列
    DMatrix testMat = new DMatrix(testInput, 1, 41, Float.NaN);
    float[][] predicts = booster.predict(testMat);

XGBoost on Spark源码中缺失值引入的不稳定问题

然而，事情并没有这么简单。

Spark ML中还有隐藏的缺失值处理逻辑：SparseVector，即稀疏向量。

SparseVector和DenseVector都用于表示一个向量，两者之间仅仅是存储结构的不同。

其中，DenseVector就是普通的Vector存储，按序存储Vector中的每一个值。

而SparseVector是稀疏的表示，用于向量中0值非常多场景下数据的存储。

SparseVector的存储方式是：仅仅记录所有非0值，忽略掉所有0值。具体来说，用一个数组记录所有非0值的位置，另一个数组记录上述位置所对应的数值。有了上述两个数组，再加上当前向量的总长度，即可将原始的数组还原回来。

因此，对于0值非常多的一组数据，SparseVector能大幅节省存储空间。

SparseVector存储示例见下图：

如上图所示，SparseVector中不保存数组中值为0的部分，仅仅记录非0值。因此对于值为0的位置其实不占用存储空间。下述代码是Spark ML中VectorAssembler的实现代码，从代码中可见，如果数值是0，在SparseVector中是不进行记录的。

scala




    

    private[feature] def assemble(vv: Any*): Vector = {
    val indices = ArrayBuilder.make[Int]
    val values = ArrayBuilder.make[Double]
    var cur = 0
    vv.foreach {
      case v: Double =>

        //0不进行保存
        if (v != 0.0) {

          indices += cur
          values += v
        }
        cur += 1
      case vec: Vector =>
        vec.foreachActive { case (i, v) =>

          //0不进行保存
          if (v != 0.0) {

            indices += cur + i
            values += v
          }
        }
        cur += vec.size
      case null =>
        throw new SparkException("Values to assemble cannot be null.")
      case o =>
        throw new SparkException(s"$o of type ${o.getClass.getName} is not supported.")
    }
    Vectors.sparse(cur, indices.result(), values.result()).compressed
    }

不占用存储空间的值，也是某种意义上的一种缺失值。SparseVector作为Spark ML中的数组的保存格式，被所有的算法组件使用，包括XGBoost on Spark。而事实上XGBoost on Spark也的确将Sparse Vector中的0值直接当作缺失值进行处理：

scala
    val instances: RDD[XGBLabeledPoint] = dataset.select(
      col($(featuresCol)),
      col($(labelCol)).cast(FloatType),
      baseMargin.cast(FloatType),
      weight.cast(FloatType)
    ).rdd.map { case Row(features: Vector, label: Float, baseMargin: Float, weight: Float) =>
      val (indices, values) = features match {

        //SparseVector格式，仅仅将非0的值放入XGBoost计算
        case v: SparseVector => (v.indices, v.values.map(_.toFloat))

        case v: DenseVector => (null, v.values.map(_.toFloat))
      }
      XGBLabeledPoint(label, indices, values, baseMargin = baseMargin, weight = weight)
    }

scala
  /**
   * Returns a vector in either dense or sparse format, whichever uses less storage.
   */
  @Since("2.0.0")
  def compressed: Vector = {
    val nnz = numNonzeros
    // A dense vector needs 8 * size + 8 bytes, while a sparse vector needs 12 * nnz + 20 bytes.
    if (1.5 * (nnz + 1.0) 
      toSparse
    } else {
      toDense
    }
  }

问题解决

scala