一.数据处理

　　org.apache.spark.mllib.util.MLUtils用于加载、保存、处理MLlib相关算法所需要的数据。其中最常用的是loadLibSVMFile，该函数用于加载LIBSVM格式的数据，返回RDD[LabeledPoint]格式的数据，用于分类、回归等算法中。

1.loadLibSVMFile

　　loadLibSVMFile用于加载指定LIBSVM格式文件的函数，源码如下：

@Since("1.0.0")
def loadLibSVMFile(
    sc: SparkContext,
    path: String,
    numFeatures: Int,
    minPartitions: Int): RDD[LabeledPoint] = {
  val parsed = parseLibSVMFile(sc, path, minPartitions)

  // Determine number of features.
  val d = if (numFeatures > 0) {
    numFeatures
  } else {
    parsed.persist(StorageLevel.MEMORY_ONLY)
    computeNumFeatures(parsed)
  }

  parsed.map { case (label, indices, values) =>
    LabeledPoint(label, Vectors.sparse(d, indices, values))
  }
}

　　其中最主要的功能之一就是根据文件路径加载数据，转化为RDD：

private[spark] def parseLibSVMFile(
    sc: SparkContext,
    path: String,
    minPartitions: Int): RDD[(Double, Array[Int], Array[Double])] = {
  sc.textFile(path, minPartitions)
    .map(_.trim)
    .filter(line => !(line.isEmpty || line.startsWith("#")))
    .map(parseLibSVMRecord)
}

　　最后使用map算子转化为LabeledPoint类型的RDD：

parsed.map { case (label, indices, values) =>

LabeledPoint(label, Vectors.sparse(d, indices, values))

}

2.saveAsLibSVMFile

　　saveAsLibSVMFile是将LIBSVM格式的数据保存到指定的文件中，源码如下：

3.appendBias

　　appendBias是对向量增加偏置项，用于分类和回归算法中。源码如下：

@Since("1.0.0")
def appendBias(vector: Vector): Vector = {
  vector match {
    case dv: DenseVector =>
      val inputValues = dv.values
      val inputLength = inputValues.length
      val outputValues = Array.ofDim[Double](inputLength + 1)
      System.arraycopy(inputValues, 0, outputValues, 0, inputLength)
      outputValues(inputLength) = 1.0
      Vectors.dense(outputValues)
    case sv: SparseVector =>
      val inputValues = sv.values
      val inputIndices = sv.indices
      val inputValuesLength = inputValues.length
      val dim = sv.size
      val outputValues = Array.ofDim[Double](inputValuesLength + 1)
      val outputIndices = Array.ofDim[Int](inputValuesLength + 1)
      System.arraycopy(inputValues, 0, outputValues, 0, inputValuesLength)
      System.arraycopy(inputIndices, 0, outputIndices, 0, inputValuesLength)
      outputValues(inputValuesLength) = 1.0
      outputIndices(inputValuesLength) = dim
      Vectors.sparse(dim + 1, outputIndices, outputValues)
    case _ => throw new IllegalArgumentException(s"Do not support vector type ${vector.getClass}")
  }
}

　　主要是匹配稀疏向量和稠密向量，否则抛出类型异常！

4.fastSquaredDistance

　　fastSquaredDistance是一种快速计算向量距离的函数，主要用于KMeans聚类算法中。它会先计算出一个精度。计算规则如下：precision = (2.0 * EPSILON * sumSquaredNorm) / (normDiff * normDiff + EPSILON)。如果在精度满足条件的情况下，可以使用欧氏距离euclideanDIst = sumSquaredNorm - 2.0 * vl.dot(v2)，其中

　　如果精度不够，则按照正规的欧式距离计算规则计算：

源码如下：

private[mllib] def fastSquaredDistance(
    v1: Vector,
    norm1: Double,
    v2: Vector,
    norm2: Double,
    precision: Double = 1e-6): Double = {
  val n = v1.size
  require(v2.size == n)
  require(norm1 >= 0.0 && norm2 >= 0.0)
  val sumSquaredNorm = norm1 * norm1 + norm2 * norm2
  val normDiff = norm1 - norm2
  var sqDist = 0.0
  val precisionBound1 = 2.0 * EPSILON * sumSquaredNorm / (normDiff * normDiff + EPSILON)
  if (precisionBound1 < precision) {
    sqDist = sumSquaredNorm - 2.0 * dot(v1, v2)
  } else if (v1.isInstanceOf[SparseVector] || v2.isInstanceOf[SparseVector]) {
    val dotValue = dot(v1, v2)
    sqDist = math.max(sumSquaredNorm - 2.0 * dotValue, 0.0)
    val precisionBound2 = EPSILON * (sumSquaredNorm + 2.0 * math.abs(dotValue)) /
      (sqDist + EPSILON)
    if (precisionBound2 > precision) {
      sqDist = Vectors.sqdist(v1, v2)
    }
  } else {
    sqDist = Vectors.sqdist(v1, v2)
  }
  sqDist
}

二.生成样本数据

1.generateKMeansRDD

　　generateKMeansRDD用于生成KMeans的训练样本数据，格式为RDD[Array[Double]]，其参数如下：

• sc ： SparkContext
• numPoints ：RDD中包含的数据量
• k ：聚类个数
• d ：数据维度
• r ：初始中心点分布的缩放因子
• numPartitions ：RDD的分区数

　　源码如下：

@Since("0.8.0")
def generateKMeansRDD(
    sc: SparkContext,
    numPoints: Int,
    k: Int,
    d: Int,
    r: Double,
    numPartitions: Int = 2)
  : RDD[Array[Double]] =
{
  // First, generate some centers
  val rand = new Random(42)
  val centers = Array.fill(k)(Array.fill(d)(rand.nextGaussian() * r))
  // Then generate points around each center
  sc.parallelize(0 until numPoints, numPartitions).map { idx =>
    val center = centers(idx % k)
    val rand2 = new Random(42 + idx)
    Array.tabulate(d)(i => center(i) + rand2.nextGaussian())
  }
}

2.generateLinearRDD

　　generateLinearRDD函数用于生成线性回归算法的训练样本数据，格式为RDD[LabeledPoint]，参数如下：

• sc ：SparkContext
• nexamples ：RDD中的数据量
• nfeatures ：样本的特征数
• eps ：Epsioon因子
• nparts ：RDD分区数

　　源码如下：

@Since("0.8.0")
def generateLinearRDD(
    sc: SparkContext,
    nexamples: Int,
    nfeatures: Int,
    eps: Double,
    nparts: Int = 2,
    intercept: Double = 0.0): RDD[LabeledPoint] = {
  val random = new Random(42)
  // Random values distributed uniformly in [-0.5, 0.5]
  val w = Array.fill(nfeatures)(random.nextDouble() - 0.5)

  val data: RDD[LabeledPoint] = sc.parallelize(0 until nparts, nparts).flatMap { p =>
    val seed = 42 + p
    val examplesInPartition = nexamples / nparts
    generateLinearInput(intercept, w.toArray, examplesInPartition, seed, eps)
  }
  data
}

3.generateLogisticRDD

　　generateLogisticRDD函数用于生成逻辑回归算法的训练样本数据，格式为RDD[LabeledPoint]，参数如下：

• sc ：SparkContext
• nexamples ：RDD中的数据量
• nfeatures ：样本的特征数
• eps ：Epsioon因子
• nparts ：RDD分区数
• probOne ：标签1的概率

　　源码如下：

@Since("0.8.0")
def generateLogisticRDD(
  sc: SparkContext,
  nexamples: Int,
  nfeatures: Int,
  eps: Double,
  nparts: Int = 2,
  probOne: Double = 0.5): RDD[LabeledPoint] = {
  val data = sc.parallelize(0 until nexamples, nparts).map { idx =>
    val rnd = new Random(42 + idx)

    val y = if (idx % 2 == 0) 0.0 else 1.0
    val x = Array.fill[Double](nfeatures) {
      rnd.nextGaussian() + (y * eps)
    }
    LabeledPoint(y, Vectors.dense(x))
  }
  data
}

4.其它数据生成

　　SVM样本生成方法：

SVMGenerator <master> <output_dir> [num_examples][num_features][num_partitions]

　　参数如下：

• master ：集群Master地址
• output_dir ：样本输出路径
• num_examples ：样本数量
• num_features ：样本特征数量
• num_partitions ：样本RDD分区数

　　源码如下：

@DeveloperApi
@Since("0.8.0")
object SVMDataGenerator {

  @Since("0.8.0")
  def main(args: Array[String]) {
    val sparkMaster: String = args(0)
    val outputPath: String = args(1)
    val nexamples: Int = if (args.length > 2) args(2).toInt else 1000
    val nfeatures: Int = if (args.length > 3) args(3).toInt else 2
    val parts: Int = if (args.length > 4) args(4).toInt else 2

    val sc = new SparkContext(sparkMaster, "SVMGenerator")

    val globalRnd = new Random(94720)
    val trueWeights = Array.fill[Double](nfeatures)(globalRnd.nextGaussian())

    val data: RDD[LabeledPoint] = sc.parallelize(0 until nexamples, parts).map { idx =>
      val rnd = new Random(42 + idx)

      val x = Array.fill[Double](nfeatures) {
        rnd.nextDouble() * 2.0 - 1.0
      }
      val yD = blas.ddot(trueWeights.length, x, 1, trueWeights, 1) + rnd.nextGaussian() * 0.1
      val y = if (yD < 0) 0.0 else 1.0
      LabeledPoint(y, Vectors.dense(x))
    }

    data.saveAsTextFile(outputPath)
    sc.stop()
  }
}

　　MFD样本生成方法：

MFDataGenerator <master> <outputDir> [m][n][rank][trainSampFact][noise][sigma][test][testSampFact]

　　参数如下：

• master ：集群Master地址
• outputDir ：样本输出路径
• m ：矩阵行数
• n ：矩阵列数
• rank ：矩阵秩
• trainSampFact ：采样因子
• noise ：是否增加噪音
• sigma ：高斯噪音标准差
• test ：是否测试
• testSampFact ：测试比例

　　源码如下：

@DeveloperApi
@Since("0.8.0")
object MFDataGenerator {
  @Since("0.8.0")
  def main(args: Array[String]) {
    val sparkMaster: String = args(0)
    val outputPath: String = args(1)
    val m: Int = if (args.length > 2) args(2).toInt else 100
    val n: Int = if (args.length > 3) args(3).toInt else 100
    val rank: Int = if (args.length > 4) args(4).toInt else 10
    val trainSampFact: Double = if (args.length > 5) args(5).toDouble else 1.0
    val noise: Boolean = if (args.length > 6) args(6).toBoolean else false
    val sigma: Double = if (args.length > 7) args(7).toDouble else 0.1
    val test: Boolean = if (args.length > 8) args(8).toBoolean else false
    val testSampFact: Double = if (args.length > 9) args(9).toDouble else 0.1

    val sc = new SparkContext(sparkMaster, "MFDataGenerator")

    val random = new ju.Random(42L)

    val A = DenseMatrix.randn(m, rank, random)
    val B = DenseMatrix.randn(rank, n, random)
    val z = 1 / math.sqrt(rank)
    val fullData = DenseMatrix.zeros(m, n)
    BLAS.gemm(z, A, B, 1.0, fullData)

    val df = rank * (m + n - rank)
    val sampSize = math.min(math.round(trainSampFact * df), math.round(.99 * m * n)).toInt
    val rand = new Random()
    val mn = m * n
    val shuffled = rand.shuffle((0 until mn).toList)

    val omega = shuffled.slice(0, sampSize)
    val ordered = omega.sortWith(_ < _).toArray
    val trainData: RDD[(Int, Int, Double)] = sc.parallelize(ordered)
      .map(x => (x % m, x / m, fullData.values(x)))

    // optionally add gaussian noise
    if (noise) {
      trainData.map(x => (x._1, x._2, x._3 + rand.nextGaussian * sigma))
    }

    trainData.map(x => x._1 + "," + x._2 + "," + x._3).saveAsTextFile(outputPath)

    // optionally generate testing data
    if (test) {
      val testSampSize = math.min(math.round(sampSize * testSampFact).toInt, mn - sampSize)
      val testOmega = shuffled.slice(sampSize, sampSize + testSampSize)
      val testOrdered = testOmega.sortWith(_ < _).toArray
      val testData: RDD[(Int, Int, Double)] = sc.parallelize(testOrdered)
        .map(x => (x % m, x / m, fullData.values(x)))
      testData.map(x => x._1 + "," + x._2 + "," + x._3).saveAsTextFile(outputPath)
    }
    sc.stop()
  }
}

　　SVM样本生成方法和MFD样本生成方法都是探索类型的API，因此不建议使用！