Parse libsvm data for spark MLlib
LibSVM data format is widely used in Machine Learning. Spark MLlib is a powerful tool to train large scale machine learning models. If your data is well formatted in LibSVM, it is straightforward to use the loadLibSVMFile method to transfer your data into an Rdd.
val data = MLUtils.loadLibSVMFile(sc, "data/mllib/sample_libsvm_data.txt")However, in certain cases, your data is not well formatted in LibSVM. For example, you may have different models, and each model has its own labeled data. Suppose your data is stored into HDFS, and each line looks like this: (model_key, training_instance_in_livsvm_format).
In this case, you can store the data by model_key, so each model_key has its own data folder. Another method is to parse the data yourself.
The following code shows how to parse libsvm data so that it can be used to train a model using Spark MLlib.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
def parseLibSVMRecord(line: String, prob:Double = 1): (Double, Array[Int], Array[Double]) = { val itemsAll = line.split(' ') val label = if (itemsAll.head.toDouble == -1) { 0 } else {1} val items= itemsAll.tail.sortWith{case (a, b) => a.split(":")(0).toInt < b.split(":")(0).toInt} val (indices, values) = items.filter(_.nonEmpty).sample(false,.01, 12345).map { item => val indexAndValue = item.split(':') val index = indexAndValue(0).toInt //- 1 // Convert 1-based indices to 0-based. val value = indexAndValue(1).toDouble (index, value) }.unzip // check if indices are one-based and in ascending order var previous = -1 var i = 0 val indicesLength = indices.length while (i < indicesLength) { val current = indices(i) require(current > previous, s"indices should be one-based and in ascending order;" + " found current=$current, previous=$previous; line=\"$line\"") previous = current i += 1 } (label, indices.toArray, values.toArray) } |
Suppose we load the data using sc.textFile(), then parse it into two parts: (model_key:String, libsvm_data_line: String ).
|
1 2 3 4 5 6 |
val rdd = sc.textFile(input) //each line like this: (model_key ^A train_instance_in_libsvm_format) val data = rdd.map(line => line.trim.split('\u0001')). filter(line => !(line.isEmpty || line.startsWith("#"))).map(ary => (ary(0), ary(1))) val keys = data.map(_._1).distinct.collect // these are the model keys |
Now we can get the train data based on a model key and parse the libsvm data into RDD[LabeledPoint].
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 |
def computeNumFeatures(rdd: RDD[(Double, Array[Int], Array[Double])]): Int = { rdd.map { case (label, indices, values) => indices.lastOption.getOrElse(0) }.reduce(math.max) + 1 } def get_train_data(rdd:RDD[(String, String)], key:String, numFeatures:Int): RDD[LabeledPoint] ={ val mrdd = rdd.filter(_._1 == key) val parsed = mrdd.map{case (mk, v) => v}.map(parseLibSVMRecord) // Determine number of features. val d = if (numFeatures > 0) { numFeatures } else { parsed.persist(StorageLevel.MEMORY_ONLY) computeNumFeatures(parsed) } val res = parsed.map { case (label, indices, values) => LabeledPoint(label, Vectors.sparse(d, indices, values)) } res } |
